async.c File Reference

#include "postgres.h"
#include <limits.h>
#include <unistd.h>
#include <signal.h>
#include "access/parallel.h"
#include "access/slru.h"
#include "access/transam.h"
#include "access/xact.h"
#include "catalog/pg_database.h"
#include "commands/async.h"
#include "common/hashfn.h"
#include "funcapi.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "storage/sinval.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/ps_status.h"
#include "utils/snapmgr.h"
#include "utils/timestamp.h"

Include dependency graph for async.c:

Go to the source code of this file.

Data Structures
struct	AsyncQueueEntry
struct	QueuePosition
struct	QueueBackendStatus
struct	AsyncQueueControl
struct	ListenAction
struct	ActionList
struct	Notification
struct	NotificationList
struct	NotificationHash

Macros
#define	NOTIFY_PAYLOAD_MAX_LENGTH   (BLCKSZ - NAMEDATALEN - 128)
#define	QUEUEALIGN(len)   INTALIGN(len)
#define	AsyncQueueEntryEmptySize   (offsetof(AsyncQueueEntry, data) + 2)
#define	QUEUE_POS_PAGE(x)   ((x).page)
#define	QUEUE_POS_OFFSET(x)   ((x).offset)
#define	SET_QUEUE_POS(x, y, z)
#define	QUEUE_POS_EQUAL(x, y)   ((x).page == (y).page && (x).offset == (y).offset)
#define	QUEUE_POS_IS_ZERO(x)   ((x).page == 0 && (x).offset == 0)
#define	QUEUE_POS_MIN(x, y)
#define	QUEUE_POS_MAX(x, y)
#define	QUEUE_CLEANUP_DELAY   4
#define	QUEUE_HEAD   (asyncQueueControl->head)
#define	QUEUE_TAIL   (asyncQueueControl->tail)
#define	QUEUE_STOP_PAGE   (asyncQueueControl->stopPage)
#define	QUEUE_FIRST_LISTENER   (asyncQueueControl->firstListener)
#define	QUEUE_BACKEND_PID(i)   (asyncQueueControl->backend[i].pid)
#define	QUEUE_BACKEND_DBOID(i)   (asyncQueueControl->backend[i].dboid)
#define	QUEUE_NEXT_LISTENER(i)   (asyncQueueControl->backend[i].nextListener)
#define	QUEUE_BACKEND_POS(i)   (asyncQueueControl->backend[i].pos)
#define	NotifyCtl   (&NotifyCtlData)
#define	QUEUE_PAGESIZE   BLCKSZ
#define	QUEUE_FULL_WARN_INTERVAL   5000 /* warn at most once every 5s */
#define	QUEUE_MAX_PAGE   (SLRU_PAGES_PER_SEGMENT * 0x10000 - 1)
#define	MIN_HASHABLE_NOTIFIES   16 /* threshold to build hashtab */

Typedefs
typedef struct AsyncQueueEntry	AsyncQueueEntry
typedef struct QueuePosition	QueuePosition
typedef struct QueueBackendStatus	QueueBackendStatus
typedef struct AsyncQueueControl	AsyncQueueControl
typedef struct ActionList	ActionList
typedef struct Notification	Notification
typedef struct NotificationList	NotificationList
typedef struct NotificationHash	NotificationHash

Enumerations
enum	ListenActionKind { LISTEN_LISTEN, LISTEN_UNLISTEN, LISTEN_UNLISTEN_ALL }

Functions
static int	asyncQueuePageDiff (int p, int q)
static bool	asyncQueuePagePrecedes (int p, int q)
static void	queue_listen (ListenActionKind action, const char *channel)
static void	Async_UnlistenOnExit (int code, Datum arg)
static void	Exec_ListenPreCommit (void)
static void	Exec_ListenCommit (const char *channel)
static void	Exec_UnlistenCommit (const char *channel)
static void	Exec_UnlistenAllCommit (void)
static bool	IsListeningOn (const char *channel)
static void	asyncQueueUnregister (void)
static bool	asyncQueueIsFull (void)
static bool	asyncQueueAdvance (volatile QueuePosition *position, int entryLength)
static void	asyncQueueNotificationToEntry (Notification *n, AsyncQueueEntry *qe)
static ListCell *	asyncQueueAddEntries (ListCell *nextNotify)
static double	asyncQueueUsage (void)
static void	asyncQueueFillWarning (void)
static void	SignalBackends (void)
static void	asyncQueueReadAllNotifications (void)
static bool	asyncQueueProcessPageEntries (volatile QueuePosition *current, QueuePosition stop, char *page_buffer, Snapshot snapshot)
static void	asyncQueueAdvanceTail (void)
static void	ProcessIncomingNotify (void)
static bool	AsyncExistsPendingNotify (Notification *n)
static void	AddEventToPendingNotifies (Notification *n)
static uint32	notification_hash (const void *key, Size keysize)
static int	notification_match (const void *key1, const void *key2, Size keysize)
static void	ClearPendingActionsAndNotifies (void)
Size	AsyncShmemSize (void)
void	AsyncShmemInit (void)
Datum	pg_notify (PG_FUNCTION_ARGS)
void	Async_Notify (const char *channel, const char *payload)
void	Async_Listen (const char *channel)
void	Async_Unlisten (const char *channel)
void	Async_UnlistenAll (void)
Datum	pg_listening_channels (PG_FUNCTION_ARGS)
void	AtPrepare_Notify (void)
void	PreCommit_Notify (void)
void	AtCommit_Notify (void)
void	ProcessCompletedNotifies (void)
Datum	pg_notification_queue_usage (PG_FUNCTION_ARGS)
void	AtAbort_Notify (void)
void	AtSubCommit_Notify (void)
void	AtSubAbort_Notify (void)
void	HandleNotifyInterrupt (void)
void	ProcessNotifyInterrupt (void)
void	NotifyMyFrontEnd (const char *channel, const char *payload, int32 srcPid)

Variables
static AsyncQueueControl *	asyncQueueControl
static SlruCtlData	NotifyCtlData
static List *	listenChannels = NIL
static ActionList *	pendingActions = NULL
static NotificationList *	pendingNotifies = NULL
volatile sig_atomic_t	notifyInterruptPending = false
static bool	unlistenExitRegistered = false
static bool	amRegisteredListener = false
static bool	backendHasSentNotifications = false
static bool	backendTryAdvanceTail = false
bool	Trace_notify = false

Macro Definition Documentation

AsyncQueueEntryEmptySize

#define AsyncQueueEntryEmptySize   (offsetof(AsyncQueueEntry, data) + 2)

Definition at line 182 of file async.c.

Referenced by asyncQueueAdvance(), and asyncQueueNotificationToEntry().

MIN_HASHABLE_NOTIFIES

#define MIN_HASHABLE_NOTIFIES   16 /* threshold to build hashtab */

Definition at line 405 of file async.c.

Referenced by AddEventToPendingNotifies().

NOTIFY_PAYLOAD_MAX_LENGTH

#define NOTIFY_PAYLOAD_MAX_LENGTH   (BLCKSZ - NAMEDATALEN - 128)

Definition at line 156 of file async.c.

Referenced by Async_Notify(), and asyncQueueNotificationToEntry().

NotifyCtl

#define NotifyCtl   (&NotifyCtlData)

Definition at line 303 of file async.c.

Referenced by asyncQueueAddEntries(), asyncQueueAdvanceTail(), asyncQueueReadAllNotifications(), and AsyncShmemInit().

QUEUE_BACKEND_DBOID

#define QUEUE_BACKEND_DBOID

(

i

)

(asyncQueueControl->backend[i].dboid)

Definition at line 294 of file async.c.

Referenced by asyncQueueUnregister(), AsyncShmemInit(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_BACKEND_PID

#define QUEUE_BACKEND_PID

(

i

)

(asyncQueueControl->backend[i].pid)

Definition at line 293 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueFillWarning(), asyncQueueReadAllNotifications(), asyncQueueUnregister(), AsyncShmemInit(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_BACKEND_POS

#define QUEUE_BACKEND_POS

(

i

)

(asyncQueueControl->backend[i].pos)

Definition at line 296 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueFillWarning(), asyncQueueReadAllNotifications(), AsyncShmemInit(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_CLEANUP_DELAY

#define QUEUE_CLEANUP_DELAY   4

Definition at line 231 of file async.c.

Referenced by asyncQueueAddEntries(), and SignalBackends().

QUEUE_FIRST_LISTENER

#define QUEUE_FIRST_LISTENER   (asyncQueueControl->firstListener)

Definition at line 292 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueFillWarning(), asyncQueueUnregister(), AsyncShmemInit(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_FULL_WARN_INTERVAL

#define QUEUE_FULL_WARN_INTERVAL   5000 /* warn at most once every 5s */

Definition at line 305 of file async.c.

Referenced by asyncQueueFillWarning().

QUEUE_HEAD

#define QUEUE_HEAD   (asyncQueueControl->head)

Definition at line 289 of file async.c.

Referenced by asyncQueueAddEntries(), asyncQueueAdvanceTail(), asyncQueueFillWarning(), asyncQueueIsFull(), asyncQueueReadAllNotifications(), asyncQueueUsage(), AsyncShmemInit(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_MAX_PAGE

#define QUEUE_MAX_PAGE   (SLRU_PAGES_PER_SEGMENT * 0x10000 - 1)

Definition at line 321 of file async.c.

Referenced by asyncQueueAdvance(), asyncQueueIsFull(), asyncQueuePageDiff(), and asyncQueueUsage().

QUEUE_NEXT_LISTENER

#define QUEUE_NEXT_LISTENER

(

i

)

(asyncQueueControl->backend[i].nextListener)

Definition at line 295 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueFillWarning(), asyncQueueUnregister(), AsyncShmemInit(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_PAGESIZE

#define QUEUE_PAGESIZE   BLCKSZ

Definition at line 304 of file async.c.

Referenced by asyncQueueAddEntries(), asyncQueueAdvance(), and asyncQueueReadAllNotifications().

QUEUE_POS_EQUAL

#define QUEUE_POS_EQUAL	(		x,
			y
	)		((x).page == (y).page && (x).offset == (y).offset)

Definition at line 202 of file async.c.

Referenced by asyncQueueFillWarning(), asyncQueueProcessPageEntries(), asyncQueueReadAllNotifications(), Exec_ListenPreCommit(), and SignalBackends().

QUEUE_POS_IS_ZERO

#define QUEUE_POS_IS_ZERO

(

x

)

((x).page == 0 && (x).offset == 0)

Definition at line 205 of file async.c.

Referenced by asyncQueueAddEntries().

QUEUE_POS_MAX

#define QUEUE_POS_MAX	(		x,
			y
	)

Value:

(asyncQueuePagePrecedes((x).page, (y).page) ? (y) : \
     (x).page != (y).page ? (x) : \
     (x).offset > (y).offset ? (x) : (y))

Definition at line 215 of file async.c.

Referenced by Exec_ListenPreCommit().

QUEUE_POS_MIN

#define QUEUE_POS_MIN	(		x,
			y
	)

Value:

(asyncQueuePagePrecedes((x).page, (y).page) ? (x) : \
     (x).page != (y).page ? (y) : \
     (x).offset < (y).offset ? (x) : (y))

Definition at line 209 of file async.c.

Referenced by asyncQueueAdvanceTail(), and asyncQueueFillWarning().

QUEUE_POS_OFFSET

#define QUEUE_POS_OFFSET

(

x

)

((x).offset)

Definition at line 194 of file async.c.

Referenced by asyncQueueAddEntries(), asyncQueueAdvance(), asyncQueueProcessPageEntries(), and asyncQueueReadAllNotifications().

QUEUE_POS_PAGE

#define QUEUE_POS_PAGE

(

x

)

((x).page)

Definition at line 193 of file async.c.

Referenced by asyncQueueAddEntries(), asyncQueueAdvance(), asyncQueueAdvanceTail(), asyncQueueIsFull(), asyncQueueReadAllNotifications(), asyncQueueUsage(), and SignalBackends().

QUEUE_STOP_PAGE

#define QUEUE_STOP_PAGE   (asyncQueueControl->stopPage)

Definition at line 291 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueIsFull(), and AsyncShmemInit().

QUEUE_TAIL

#define QUEUE_TAIL   (asyncQueueControl->tail)

Definition at line 290 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueUsage(), AsyncShmemInit(), and Exec_ListenPreCommit().

QUEUEALIGN

#define QUEUEALIGN

(

len

)

INTALIGN(len)

Definition at line 180 of file async.c.

Referenced by asyncQueueAdvance(), and asyncQueueNotificationToEntry().

SET_QUEUE_POS

#define SET_QUEUE_POS	(		x,
			y,
			z
	)

Value:

do { \
        (x).page = (y); \
        (x).offset = (z); \
    } while (0)

Definition at line 196 of file async.c.

Referenced by asyncQueueAdvance(), and AsyncShmemInit().

Typedef Documentation

ActionList

typedef struct ActionList ActionList

AsyncQueueControl

typedef struct AsyncQueueControl AsyncQueueControl

AsyncQueueEntry

typedef struct AsyncQueueEntry AsyncQueueEntry

Notification

typedef struct Notification Notification

NotificationHash

typedef struct NotificationHash NotificationHash

NotificationList

typedef struct NotificationList NotificationList

QueueBackendStatus

typedef struct QueueBackendStatus QueueBackendStatus

QueuePosition

typedef struct QueuePosition QueuePosition

Enumeration Type Documentation

ListenActionKind

enum ListenActionKind

Enumerator
LISTEN_LISTEN
LISTEN_UNLISTEN
LISTEN_UNLISTEN_ALL

Definition at line 340 of file async.c.

{
    LISTEN_LISTEN,
    LISTEN_UNLISTEN,
    LISTEN_UNLISTEN_ALL
} ListenActionKind;

Function Documentation

AddEventToPendingNotifies()

static void AddEventToPendingNotifies ( Notification *  n )

static

Definition at line 2371 of file async.c.

References Assert, CurTransactionContext, HASHCTL::entrysize, NotificationHash::event, NotificationList::events, HASHCTL::hash, HASH_COMPARE, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, HASH_FUNCTION, hash_search(), NotificationList::hashtab, HASHCTL::hcxt, HASHCTL::keysize, lappend(), lfirst, list_length(), HASHCTL::match, MIN_HASHABLE_NOTIFIES, NIL, notification_hash(), and notification_match().

Referenced by Async_Notify(), and AtSubCommit_Notify().

{
    Assert(pendingNotifies->events != NIL);

    /* Create the hash table if it's time to */
    if (list_length(pendingNotifies->events) >= MIN_HASHABLE_NOTIFIES &&
        pendingNotifies->hashtab == NULL)
    {
        HASHCTL     hash_ctl;
        ListCell   *l;

        /* Create the hash table */
        hash_ctl.keysize = sizeof(Notification *);
        hash_ctl.entrysize = sizeof(NotificationHash);
        hash_ctl.hash = notification_hash;
        hash_ctl.match = notification_match;
        hash_ctl.hcxt = CurTransactionContext;
        pendingNotifies->hashtab =
            hash_create("Pending Notifies",
                        256L,
                        &hash_ctl,
                        HASH_ELEM | HASH_FUNCTION | HASH_COMPARE | HASH_CONTEXT);

        /* Insert all the already-existing events */
        foreach(l, pendingNotifies->events)
        {
            Notification *oldn = (Notification *) lfirst(l);
            NotificationHash *hentry;
            bool        found;

            hentry = (NotificationHash *) hash_search(pendingNotifies->hashtab,
                                                      &oldn,
                                                      HASH_ENTER,
                                                      &found);
            Assert(!found);
            hentry->event = oldn;
        }
    }

    /* Add new event to the list, in order */
    pendingNotifies->events = lappend(pendingNotifies->events, n);

    /* Add event to the hash table if needed */
    if (pendingNotifies->hashtab != NULL)
    {
        NotificationHash *hentry;
        bool        found;

        hentry = (NotificationHash *) hash_search(pendingNotifies->hashtab,
                                                  &n,
                                                  HASH_ENTER,
                                                  &found);
        Assert(!found);
        hentry->event = n;
    }
}

Async_Listen()

void Async_Listen

(

const char *

channel

)

Definition at line 765 of file async.c.

References DEBUG1, elog, LISTEN_LISTEN, MyProcPid, queue_listen(), and Trace_notify.

Referenced by standard_ProcessUtility().

{
    if (Trace_notify)
        elog(DEBUG1, "Async_Listen(%s,%d)", channel, MyProcPid);

    queue_listen(LISTEN_LISTEN, channel);
}

Async_Notify()

void Async_Notify	(	const char *	channel,
		const char *	payload
	)

Definition at line 618 of file async.c.

References AddEventToPendingNotifies(), AsyncExistsPendingNotify(), Notification::channel_len, CurTransactionContext, AsyncQueueEntry::data, Notification::data, DEBUG1, elog, ereport, errcode(), errmsg(), ERROR, NotificationList::events, GetCurrentTransactionNestLevel(), NotificationList::hashtab, IsParallelWorker, list_make1, MemoryContextAlloc(), MemoryContextSwitchTo(), NAMEDATALEN, NotificationList::nestingLevel, NOTIFY_PAYLOAD_MAX_LENGTH, offsetof, palloc(), Notification::payload_len, pendingNotifies, pfree(), TopTransactionContext, Trace_notify, and NotificationList::upper.

Referenced by pg_notify(), standard_ProcessUtility(), and triggered_change_notification().

{
    int         my_level = GetCurrentTransactionNestLevel();
    size_t      channel_len;
    size_t      payload_len;
    Notification *n;
    MemoryContext oldcontext;

    if (IsParallelWorker())
        elog(ERROR, "cannot send notifications from a parallel worker");

    if (Trace_notify)
        elog(DEBUG1, "Async_Notify(%s)", channel);

    channel_len = channel ? strlen(channel) : 0;
    payload_len = payload ? strlen(payload) : 0;

    /* a channel name must be specified */
    if (channel_len == 0)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("channel name cannot be empty")));

    /* enforce length limits */
    if (channel_len >= NAMEDATALEN)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("channel name too long")));

    if (payload_len >= NOTIFY_PAYLOAD_MAX_LENGTH)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("payload string too long")));

    /*
     * We must construct the Notification entry, even if we end up not using
     * it, in order to compare it cheaply to existing list entries.
     *
     * The notification list needs to live until end of transaction, so store
     * it in the transaction context.
     */
    oldcontext = MemoryContextSwitchTo(CurTransactionContext);

    n = (Notification *) palloc(offsetof(Notification, data) +
                                channel_len + payload_len + 2);
    n->channel_len = channel_len;
    n->payload_len = payload_len;
    strcpy(n->data, channel);
    if (payload)
        strcpy(n->data + channel_len + 1, payload);
    else
        n->data[channel_len + 1] = '\0';

    if (pendingNotifies == NULL || my_level > pendingNotifies->nestingLevel)
    {
        NotificationList *notifies;

        /*
         * First notify event in current (sub)xact. Note that we allocate the
         * NotificationList in TopTransactionContext; the nestingLevel might
         * get changed later by AtSubCommit_Notify.
         */
        notifies = (NotificationList *)
            MemoryContextAlloc(TopTransactionContext,
                               sizeof(NotificationList));
        notifies->nestingLevel = my_level;
        notifies->events = list_make1(n);
        /* We certainly don't need a hashtable yet */
        notifies->hashtab = NULL;
        notifies->upper = pendingNotifies;
        pendingNotifies = notifies;
    }
    else
    {
        /* Now check for duplicates */
        if (AsyncExistsPendingNotify(n))
        {
            /* It's a dup, so forget it */
            pfree(n);
            MemoryContextSwitchTo(oldcontext);
            return;
        }

        /* Append more events to existing list */
        AddEventToPendingNotifies(n);
    }

    MemoryContextSwitchTo(oldcontext);
}

Async_Unlisten()

void Async_Unlisten

(

const char *

channel

)

Definition at line 779 of file async.c.

References DEBUG1, elog, LISTEN_UNLISTEN, MyProcPid, queue_listen(), Trace_notify, and unlistenExitRegistered.

Referenced by standard_ProcessUtility().

{
    if (Trace_notify)
        elog(DEBUG1, "Async_Unlisten(%s,%d)", channel, MyProcPid);

    /* If we couldn't possibly be listening, no need to queue anything */
    if (pendingActions == NULL && !unlistenExitRegistered)
        return;

    queue_listen(LISTEN_UNLISTEN, channel);
}

Async_UnlistenAll()

void Async_UnlistenAll

(

void

)

Definition at line 797 of file async.c.

References DEBUG1, elog, LISTEN_UNLISTEN_ALL, MyProcPid, queue_listen(), Trace_notify, and unlistenExitRegistered.

Referenced by DiscardAll(), and standard_ProcessUtility().

{
    if (Trace_notify)
        elog(DEBUG1, "Async_UnlistenAll(%d)", MyProcPid);

    /* If we couldn't possibly be listening, no need to queue anything */
    if (pendingActions == NULL && !unlistenExitRegistered)
        return;

    queue_listen(LISTEN_UNLISTEN_ALL, "");
}

Async_UnlistenOnExit()

static void Async_UnlistenOnExit ( int  code, Datum  arg  )

static

Definition at line 850 of file async.c.

References asyncQueueUnregister(), and Exec_UnlistenAllCommit().

Referenced by Exec_ListenPreCommit().

{
    Exec_UnlistenAllCommit();
    asyncQueueUnregister();
}

AsyncExistsPendingNotify()

static bool AsyncExistsPendingNotify ( Notification *  n )

static

Definition at line 2330 of file async.c.

References Notification::channel_len, Notification::data, NotificationList::events, HASH_FIND, hash_search(), NotificationList::hashtab, lfirst, and Notification::payload_len.

Referenced by Async_Notify(), and AtSubCommit_Notify().

{
    if (pendingNotifies == NULL)
        return false;

    if (pendingNotifies->hashtab != NULL)
    {
        /* Use the hash table to probe for a match */
        if (hash_search(pendingNotifies->hashtab,
                        &n,
                        HASH_FIND,
                        NULL))
            return true;
    }
    else
    {
        /* Must scan the event list */
        ListCell   *l;

        foreach(l, pendingNotifies->events)
        {
            Notification *oldn = (Notification *) lfirst(l);

            if (n->channel_len == oldn->channel_len &&
                n->payload_len == oldn->payload_len &&
                memcmp(n->data, oldn->data,
                       n->channel_len + n->payload_len + 2) == 0)
                return true;
        }
    }

    return false;
}

asyncQueueAddEntries()

static ListCell * asyncQueueAddEntries ( ListCell *  nextNotify )

static

Definition at line 1452 of file async.c.

References asyncQueueAdvance(), asyncQueueNotificationToEntry(), backendTryAdvanceTail, AsyncQueueEntry::data, AsyncQueueEntry::dboid, NotificationList::events, InvalidOid, InvalidTransactionId, AsyncQueueEntry::length, lfirst, lnext(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), NotifyCtl, QUEUE_CLEANUP_DELAY, QUEUE_HEAD, QUEUE_PAGESIZE, QUEUE_POS_IS_ZERO, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, SimpleLruReadPage(), and SimpleLruZeroPage().

Referenced by PreCommit_Notify().

{
    AsyncQueueEntry qe;
    QueuePosition queue_head;
    int         pageno;
    int         offset;
    int         slotno;

    /* We hold both NotifyQueueLock and NotifySLRULock during this operation */
    LWLockAcquire(NotifySLRULock, LW_EXCLUSIVE);

    /*
     * We work with a local copy of QUEUE_HEAD, which we write back to shared
     * memory upon exiting.  The reason for this is that if we have to advance
     * to a new page, SimpleLruZeroPage might fail (out of disk space, for
     * instance), and we must not advance QUEUE_HEAD if it does.  (Otherwise,
     * subsequent insertions would try to put entries into a page that slru.c
     * thinks doesn't exist yet.)  So, use a local position variable.  Note
     * that if we do fail, any already-inserted queue entries are forgotten;
     * this is okay, since they'd be useless anyway after our transaction
     * rolls back.
     */
    queue_head = QUEUE_HEAD;

    /*
     * If this is the first write since the postmaster started, we need to
     * initialize the first page of the async SLRU.  Otherwise, the current
     * page should be initialized already, so just fetch it.
     *
     * (We could also take the first path when the SLRU position has just
     * wrapped around, but re-zeroing the page is harmless in that case.)
     */
    pageno = QUEUE_POS_PAGE(queue_head);
    if (QUEUE_POS_IS_ZERO(queue_head))
        slotno = SimpleLruZeroPage(NotifyCtl, pageno);
    else
        slotno = SimpleLruReadPage(NotifyCtl, pageno, true,
                                   InvalidTransactionId);

    /* Note we mark the page dirty before writing in it */
    NotifyCtl->shared->page_dirty[slotno] = true;

    while (nextNotify != NULL)
    {
        Notification *n = (Notification *) lfirst(nextNotify);

        /* Construct a valid queue entry in local variable qe */
        asyncQueueNotificationToEntry(n, &qe);

        offset = QUEUE_POS_OFFSET(queue_head);

        /* Check whether the entry really fits on the current page */
        if (offset + qe.length <= QUEUE_PAGESIZE)
        {
            /* OK, so advance nextNotify past this item */
            nextNotify = lnext(pendingNotifies->events, nextNotify);
        }
        else
        {
            /*
             * Write a dummy entry to fill up the page. Actually readers will
             * only check dboid and since it won't match any reader's database
             * OID, they will ignore this entry and move on.
             */
            qe.length = QUEUE_PAGESIZE - offset;
            qe.dboid = InvalidOid;
            qe.data[0] = '\0';  /* empty channel */
            qe.data[1] = '\0';  /* empty payload */
        }

        /* Now copy qe into the shared buffer page */
        memcpy(NotifyCtl->shared->page_buffer[slotno] + offset,
               &qe,
               qe.length);

        /* Advance queue_head appropriately, and detect if page is full */
        if (asyncQueueAdvance(&(queue_head), qe.length))
        {
            /*
             * Page is full, so we're done here, but first fill the next page
             * with zeroes.  The reason to do this is to ensure that slru.c's
             * idea of the head page is always the same as ours, which avoids
             * boundary problems in SimpleLruTruncate.  The test in
             * asyncQueueIsFull() ensured that there is room to create this
             * page without overrunning the queue.
             */
            slotno = SimpleLruZeroPage(NotifyCtl, QUEUE_POS_PAGE(queue_head));

            /*
             * If the new page address is a multiple of QUEUE_CLEANUP_DELAY,
             * set flag to remember that we should try to advance the tail
             * pointer (we don't want to actually do that right here).
             */
            if (QUEUE_POS_PAGE(queue_head) % QUEUE_CLEANUP_DELAY == 0)
                backendTryAdvanceTail = true;

            /* And exit the loop */
            break;
        }
    }

    /* Success, so update the global QUEUE_HEAD */
    QUEUE_HEAD = queue_head;

    LWLockRelease(NotifySLRULock);

    return nextNotify;
}

asyncQueueAdvance()

static bool asyncQueueAdvance ( volatile QueuePosition *  position, int  entryLength  )

static

Definition at line 1381 of file async.c.

References Assert, AsyncQueueEntryEmptySize, QUEUE_MAX_PAGE, QUEUE_PAGESIZE, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, QUEUEALIGN, and SET_QUEUE_POS.

Referenced by asyncQueueAddEntries(), and asyncQueueProcessPageEntries().

{
    int         pageno = QUEUE_POS_PAGE(*position);
    int         offset = QUEUE_POS_OFFSET(*position);
    bool        pageJump = false;

    /*
     * Move to the next writing position: First jump over what we have just
     * written or read.
     */
    offset += entryLength;
    Assert(offset <= QUEUE_PAGESIZE);

    /*
     * In a second step check if another entry can possibly be written to the
     * page. If so, stay here, we have reached the next position. If not, then
     * we need to move on to the next page.
     */
    if (offset + QUEUEALIGN(AsyncQueueEntryEmptySize) > QUEUE_PAGESIZE)
    {
        pageno++;
        if (pageno > QUEUE_MAX_PAGE)
            pageno = 0;         /* wrap around */
        offset = 0;
        pageJump = true;
    }

    SET_QUEUE_POS(*position, pageno, offset);
    return pageJump;
}

asyncQueueAdvanceTail()

static void asyncQueueAdvanceTail ( void  )

static

Definition at line 2185 of file async.c.

References Assert, asyncQueuePagePrecedes(), i, InvalidPid, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), NotifyCtl, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_FIRST_LISTENER, QUEUE_HEAD, QUEUE_NEXT_LISTENER, QUEUE_POS_MIN, QUEUE_POS_PAGE, QUEUE_STOP_PAGE, QUEUE_TAIL, SimpleLruTruncate(), and SLRU_PAGES_PER_SEGMENT.

Referenced by pg_notification_queue_usage(), and ProcessCompletedNotifies().

{
    QueuePosition min;
    int         oldtailpage;
    int         newtailpage;
    int         boundary;

    /* Restrict task to one backend per cluster; see SimpleLruTruncate(). */
    LWLockAcquire(NotifyQueueTailLock, LW_EXCLUSIVE);

    /*
     * Compute the new tail.  Pre-v13, it's essential that QUEUE_TAIL be exact
     * (ie, exactly match at least one backend's queue position), so it must
     * be updated atomically with the actual computation.  Since v13, we could
     * get away with not doing it like that, but it seems prudent to keep it
     * so.
     *
     * Also, because incoming backends will scan forward from QUEUE_TAIL, that
     * must be advanced before we can truncate any data.  Thus, QUEUE_TAIL is
     * the logical tail, while QUEUE_STOP_PAGE is the physical tail, or oldest
     * un-truncated page.  When QUEUE_STOP_PAGE != QUEUE_POS_PAGE(QUEUE_TAIL),
     * there are pages we can truncate but haven't yet finished doing so.
     *
     * For concurrency's sake, we don't want to hold NotifyQueueLock while
     * performing SimpleLruTruncate.  This is OK because no backend will try
     * to access the pages we are in the midst of truncating.
     */
    LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
    min = QUEUE_HEAD;
    for (BackendId i = QUEUE_FIRST_LISTENER; i > 0; i = QUEUE_NEXT_LISTENER(i))
    {
        Assert(QUEUE_BACKEND_PID(i) != InvalidPid);
        min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i));
    }
    QUEUE_TAIL = min;
    oldtailpage = QUEUE_STOP_PAGE;
    LWLockRelease(NotifyQueueLock);

    /*
     * We can truncate something if the global tail advanced across an SLRU
     * segment boundary.
     *
     * XXX it might be better to truncate only once every several segments, to
     * reduce the number of directory scans.
     */
    newtailpage = QUEUE_POS_PAGE(min);
    boundary = newtailpage - (newtailpage % SLRU_PAGES_PER_SEGMENT);
    if (asyncQueuePagePrecedes(oldtailpage, boundary))
    {
        /*
         * SimpleLruTruncate() will ask for NotifySLRULock but will also
         * release the lock again.
         */
        SimpleLruTruncate(NotifyCtl, newtailpage);

        /*
         * Update QUEUE_STOP_PAGE.  This changes asyncQueueIsFull()'s verdict
         * for the segment immediately prior to the old tail, allowing fresh
         * data into that segment.
         */
        LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
        QUEUE_STOP_PAGE = newtailpage;
        LWLockRelease(NotifyQueueLock);
    }

    LWLockRelease(NotifyQueueTailLock);
}

asyncQueueFillWarning()

static void asyncQueueFillWarning ( void  )

static

Definition at line 1620 of file async.c.

References Assert, asyncQueueUsage(), ereport, errdetail(), errhint(), errmsg(), GetCurrentTimestamp(), i, InvalidPid, AsyncQueueControl::lastQueueFillWarn, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_FIRST_LISTENER, QUEUE_FULL_WARN_INTERVAL, QUEUE_HEAD, QUEUE_NEXT_LISTENER, QUEUE_POS_EQUAL, QUEUE_POS_MIN, TimestampDifferenceExceeds(), and WARNING.

Referenced by PreCommit_Notify().

{
    double      fillDegree;
    TimestampTz t;

    fillDegree = asyncQueueUsage();
    if (fillDegree < 0.5)
        return;

    t = GetCurrentTimestamp();

    if (TimestampDifferenceExceeds(asyncQueueControl->lastQueueFillWarn,
                                   t, QUEUE_FULL_WARN_INTERVAL))
    {
        QueuePosition min = QUEUE_HEAD;
        int32       minPid = InvalidPid;

        for (BackendId i = QUEUE_FIRST_LISTENER; i > 0; i = QUEUE_NEXT_LISTENER(i))
        {
            Assert(QUEUE_BACKEND_PID(i) != InvalidPid);
            min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i));
            if (QUEUE_POS_EQUAL(min, QUEUE_BACKEND_POS(i)))
                minPid = QUEUE_BACKEND_PID(i);
        }

        ereport(WARNING,
                (errmsg("NOTIFY queue is %.0f%% full", fillDegree * 100),
                 (minPid != InvalidPid ?
                  errdetail("The server process with PID %d is among those with the oldest transactions.", minPid)
                  : 0),
                 (minPid != InvalidPid ?
                  errhint("The NOTIFY queue cannot be emptied until that process ends its current transaction.")
                  : 0)));

        asyncQueueControl->lastQueueFillWarn = t;
    }
}

asyncQueueIsFull()

static bool asyncQueueIsFull ( void  )

static

Definition at line 1350 of file async.c.

References asyncQueuePagePrecedes(), QUEUE_HEAD, QUEUE_MAX_PAGE, QUEUE_POS_PAGE, QUEUE_STOP_PAGE, and SLRU_PAGES_PER_SEGMENT.

Referenced by PreCommit_Notify().

{
    int         nexthead;
    int         boundary;

    /*
     * The queue is full if creating a new head page would create a page that
     * logically precedes the current global tail pointer, ie, the head
     * pointer would wrap around compared to the tail.  We cannot create such
     * a head page for fear of confusing slru.c.  For safety we round the tail
     * pointer back to a segment boundary (truncation logic in
     * asyncQueueAdvanceTail does not do this, so doing it here is optional).
     *
     * Note that this test is *not* dependent on how much space there is on
     * the current head page.  This is necessary because asyncQueueAddEntries
     * might try to create the next head page in any case.
     */
    nexthead = QUEUE_POS_PAGE(QUEUE_HEAD) + 1;
    if (nexthead > QUEUE_MAX_PAGE)
        nexthead = 0;           /* wrap around */
    boundary = QUEUE_STOP_PAGE;
    boundary -= boundary % SLRU_PAGES_PER_SEGMENT;
    return asyncQueuePagePrecedes(nexthead, boundary);
}

asyncQueueNotificationToEntry()

static void asyncQueueNotificationToEntry ( Notification *  n, AsyncQueueEntry *  qe  )

static

Definition at line 1416 of file async.c.

References Assert, AsyncQueueEntryEmptySize, Notification::channel_len, AsyncQueueEntry::data, Notification::data, AsyncQueueEntry::dboid, GetCurrentTransactionId(), AsyncQueueEntry::length, MyDatabaseId, MyProcPid, NAMEDATALEN, NOTIFY_PAYLOAD_MAX_LENGTH, Notification::payload_len, QUEUEALIGN, AsyncQueueEntry::srcPid, and AsyncQueueEntry::xid.

Referenced by asyncQueueAddEntries().

{
    size_t      channellen = n->channel_len;
    size_t      payloadlen = n->payload_len;
    int         entryLength;

    Assert(channellen < NAMEDATALEN);
    Assert(payloadlen < NOTIFY_PAYLOAD_MAX_LENGTH);

    /* The terminators are already included in AsyncQueueEntryEmptySize */
    entryLength = AsyncQueueEntryEmptySize + payloadlen + channellen;
    entryLength = QUEUEALIGN(entryLength);
    qe->length = entryLength;
    qe->dboid = MyDatabaseId;
    qe->xid = GetCurrentTransactionId();
    qe->srcPid = MyProcPid;
    memcpy(qe->data, n->data, channellen + payloadlen + 2);
}

asyncQueuePageDiff()

static int asyncQueuePageDiff ( int  p, int  q  )

static

Definition at line 474 of file async.c.

References Assert, and QUEUE_MAX_PAGE.

Referenced by asyncQueuePagePrecedes(), and SignalBackends().

{
    int         diff;

    /*
     * We have to compare modulo (QUEUE_MAX_PAGE+1)/2.  Both inputs should be
     * in the range 0..QUEUE_MAX_PAGE.
     */
    Assert(p >= 0 && p <= QUEUE_MAX_PAGE);
    Assert(q >= 0 && q <= QUEUE_MAX_PAGE);

    diff = p - q;
    if (diff >= ((QUEUE_MAX_PAGE + 1) / 2))
        diff -= QUEUE_MAX_PAGE + 1;
    else if (diff < -((QUEUE_MAX_PAGE + 1) / 2))
        diff += QUEUE_MAX_PAGE + 1;
    return diff;
}

asyncQueuePagePrecedes()

static bool asyncQueuePagePrecedes ( int  p, int  q  )

static

Definition at line 500 of file async.c.

References asyncQueuePageDiff().

Referenced by asyncQueueAdvanceTail(), asyncQueueIsFull(), and AsyncShmemInit().

{
    return asyncQueuePageDiff(p, q) < 0;
}

asyncQueueProcessPageEntries()

static bool asyncQueueProcessPageEntries ( volatile QueuePosition *  current, QueuePosition  stop, char *  page_buffer, Snapshot  snapshot  )

static

Definition at line 2096 of file async.c.

References asyncQueueAdvance(), AsyncQueueEntry::data, AsyncQueueEntry::dboid, IsListeningOn(), AsyncQueueEntry::length, MyDatabaseId, NotifyMyFrontEnd(), QUEUE_POS_EQUAL, QUEUE_POS_OFFSET, AsyncQueueEntry::srcPid, TransactionIdDidCommit(), AsyncQueueEntry::xid, and XidInMVCCSnapshot().

Referenced by asyncQueueReadAllNotifications().

{
    bool        reachedStop = false;
    bool        reachedEndOfPage;
    AsyncQueueEntry *qe;

    do
    {
        QueuePosition thisentry = *current;

        if (QUEUE_POS_EQUAL(thisentry, stop))
            break;

        qe = (AsyncQueueEntry *) (page_buffer + QUEUE_POS_OFFSET(thisentry));

        /*
         * Advance *current over this message, possibly to the next page. As
         * noted in the comments for asyncQueueReadAllNotifications, we must
         * do this before possibly failing while processing the message.
         */
        reachedEndOfPage = asyncQueueAdvance(current, qe->length);

        /* Ignore messages destined for other databases */
        if (qe->dboid == MyDatabaseId)
        {
            if (XidInMVCCSnapshot(qe->xid, snapshot))
            {
                /*
                 * The source transaction is still in progress, so we can't
                 * process this message yet.  Break out of the loop, but first
                 * back up *current so we will reprocess the message next
                 * time.  (Note: it is unlikely but not impossible for
                 * TransactionIdDidCommit to fail, so we can't really avoid
                 * this advance-then-back-up behavior when dealing with an
                 * uncommitted message.)
                 *
                 * Note that we must test XidInMVCCSnapshot before we test
                 * TransactionIdDidCommit, else we might return a message from
                 * a transaction that is not yet visible to snapshots; compare
                 * the comments at the head of heapam_visibility.c.
                 *
                 * Also, while our own xact won't be listed in the snapshot,
                 * we need not check for TransactionIdIsCurrentTransactionId
                 * because our transaction cannot (yet) have queued any
                 * messages.
                 */
                *current = thisentry;
                reachedStop = true;
                break;
            }
            else if (TransactionIdDidCommit(qe->xid))
            {
                /* qe->data is the null-terminated channel name */
                char       *channel = qe->data;

                if (IsListeningOn(channel))
                {
                    /* payload follows channel name */
                    char       *payload = qe->data + strlen(channel) + 1;

                    NotifyMyFrontEnd(channel, payload, qe->srcPid);
                }
            }
            else
            {
                /*
                 * The source transaction aborted or crashed, so we just
                 * ignore its notifications.
                 */
            }
        }

        /* Loop back if we're not at end of page */
    } while (!reachedEndOfPage);

    if (QUEUE_POS_EQUAL(*current, stop))
        reachedStop = true;

    return reachedStop;
}

asyncQueueReadAllNotifications()

static void asyncQueueReadAllNotifications ( void  )

static

Definition at line 1931 of file async.c.

References Assert, asyncQueueProcessPageEntries(), buf, GetLatestSnapshot(), InvalidTransactionId, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyBackendId, MyProcPid, NotifyCtl, PG_END_TRY, PG_FINALLY, PG_TRY, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_HEAD, QUEUE_PAGESIZE, QUEUE_POS_EQUAL, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, RegisterSnapshot(), SimpleLruReadPage_ReadOnly(), and UnregisterSnapshot().

Referenced by Exec_ListenPreCommit(), ProcessCompletedNotifies(), and ProcessIncomingNotify().

{
    volatile QueuePosition pos;
    QueuePosition head;
    Snapshot    snapshot;

    /* page_buffer must be adequately aligned, so use a union */
    union
    {
        char        buf[QUEUE_PAGESIZE];
        AsyncQueueEntry align;
    }           page_buffer;

    /* Fetch current state */
    LWLockAcquire(NotifyQueueLock, LW_SHARED);
    /* Assert checks that we have a valid state entry */
    Assert(MyProcPid == QUEUE_BACKEND_PID(MyBackendId));
    pos = QUEUE_BACKEND_POS(MyBackendId);
    head = QUEUE_HEAD;
    LWLockRelease(NotifyQueueLock);

    if (QUEUE_POS_EQUAL(pos, head))
    {
        /* Nothing to do, we have read all notifications already. */
        return;
    }

    /*----------
     * Get snapshot we'll use to decide which xacts are still in progress.
     * This is trickier than it might seem, because of race conditions.
     * Consider the following example:
     *
     * Backend 1:                    Backend 2:
     *
     * transaction starts
     * UPDATE foo SET ...;
     * NOTIFY foo;
     * commit starts
     * queue the notify message
     *                               transaction starts
     *                               LISTEN foo;  -- first LISTEN in session
     *                               SELECT * FROM foo WHERE ...;
     * commit to clog
     *                               commit starts
     *                               add backend 2 to array of listeners
     *                               advance to queue head (this code)
     *                               commit to clog
     *
     * Transaction 2's SELECT has not seen the UPDATE's effects, since that
     * wasn't committed yet.  Ideally we'd ensure that client 2 would
     * eventually get transaction 1's notify message, but there's no way
     * to do that; until we're in the listener array, there's no guarantee
     * that the notify message doesn't get removed from the queue.
     *
     * Therefore the coding technique transaction 2 is using is unsafe:
     * applications must commit a LISTEN before inspecting database state,
     * if they want to ensure they will see notifications about subsequent
     * changes to that state.
     *
     * What we do guarantee is that we'll see all notifications from
     * transactions committing after the snapshot we take here.
     * Exec_ListenPreCommit has already added us to the listener array,
     * so no not-yet-committed messages can be removed from the queue
     * before we see them.
     *----------
     */
    snapshot = RegisterSnapshot(GetLatestSnapshot());

    /*
     * It is possible that we fail while trying to send a message to our
     * frontend (for example, because of encoding conversion failure).  If
     * that happens it is critical that we not try to send the same message
     * over and over again.  Therefore, we place a PG_TRY block here that will
     * forcibly advance our queue position before we lose control to an error.
     * (We could alternatively retake NotifyQueueLock and move the position
     * before handling each individual message, but that seems like too much
     * lock traffic.)
     */
    PG_TRY();
    {
        bool        reachedStop;

        do
        {
            int         curpage = QUEUE_POS_PAGE(pos);
            int         curoffset = QUEUE_POS_OFFSET(pos);
            int         slotno;
            int         copysize;

            /*
             * We copy the data from SLRU into a local buffer, so as to avoid
             * holding the NotifySLRULock while we are examining the entries
             * and possibly transmitting them to our frontend.  Copy only the
             * part of the page we will actually inspect.
             */
            slotno = SimpleLruReadPage_ReadOnly(NotifyCtl, curpage,
                                                InvalidTransactionId);
            if (curpage == QUEUE_POS_PAGE(head))
            {
                /* we only want to read as far as head */
                copysize = QUEUE_POS_OFFSET(head) - curoffset;
                if (copysize < 0)
                    copysize = 0;   /* just for safety */
            }
            else
            {
                /* fetch all the rest of the page */
                copysize = QUEUE_PAGESIZE - curoffset;
            }
            memcpy(page_buffer.buf + curoffset,
                   NotifyCtl->shared->page_buffer[slotno] + curoffset,
                   copysize);
            /* Release lock that we got from SimpleLruReadPage_ReadOnly() */
            LWLockRelease(NotifySLRULock);

            /*
             * Process messages up to the stop position, end of page, or an
             * uncommitted message.
             *
             * Our stop position is what we found to be the head's position
             * when we entered this function. It might have changed already.
             * But if it has, we will receive (or have already received and
             * queued) another signal and come here again.
             *
             * We are not holding NotifyQueueLock here! The queue can only
             * extend beyond the head pointer (see above) and we leave our
             * backend's pointer where it is so nobody will truncate or
             * rewrite pages under us. Especially we don't want to hold a lock
             * while sending the notifications to the frontend.
             */
            reachedStop = asyncQueueProcessPageEntries(&pos, head,
                                                       page_buffer.buf,
                                                       snapshot);
        } while (!reachedStop);
    }
    PG_FINALLY();
    {
        /* Update shared state */
        LWLockAcquire(NotifyQueueLock, LW_SHARED);
        QUEUE_BACKEND_POS(MyBackendId) = pos;
        LWLockRelease(NotifyQueueLock);
    }
    PG_END_TRY();

    /* Done with snapshot */
    UnregisterSnapshot(snapshot);
}

asyncQueueUnregister()

static void asyncQueueUnregister ( void  )

static

Definition at line 1309 of file async.c.

References amRegisteredListener, Assert, i, InvalidBackendId, InvalidOid, InvalidPid, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyBackendId, NIL, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_FIRST_LISTENER, and QUEUE_NEXT_LISTENER.

Referenced by Async_UnlistenOnExit(), AtAbort_Notify(), and AtCommit_Notify().

{
    Assert(listenChannels == NIL);  /* else caller error */

    if (!amRegisteredListener)  /* nothing to do */
        return;

    /*
     * Need exclusive lock here to manipulate list links.
     */
    LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
    /* Mark our entry as invalid */
    QUEUE_BACKEND_PID(MyBackendId) = InvalidPid;
    QUEUE_BACKEND_DBOID(MyBackendId) = InvalidOid;
    /* and remove it from the list */
    if (QUEUE_FIRST_LISTENER == MyBackendId)
        QUEUE_FIRST_LISTENER = QUEUE_NEXT_LISTENER(MyBackendId);
    else
    {
        for (BackendId i = QUEUE_FIRST_LISTENER; i > 0; i = QUEUE_NEXT_LISTENER(i))
        {
            if (QUEUE_NEXT_LISTENER(i) == MyBackendId)
            {
                QUEUE_NEXT_LISTENER(i) = QUEUE_NEXT_LISTENER(MyBackendId);
                break;
            }
        }
    }
    QUEUE_NEXT_LISTENER(MyBackendId) = InvalidBackendId;
    LWLockRelease(NotifyQueueLock);

    /* mark ourselves as no longer listed in the global array */
    amRegisteredListener = false;
}

asyncQueueUsage()

static double asyncQueueUsage ( void  )

static

Definition at line 1591 of file async.c.

References QUEUE_HEAD, QUEUE_MAX_PAGE, QUEUE_POS_PAGE, and QUEUE_TAIL.

Referenced by asyncQueueFillWarning(), and pg_notification_queue_usage().

{
    int         headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
    int         tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
    int         occupied;

    occupied = headPage - tailPage;

    if (occupied == 0)
        return (double) 0;      /* fast exit for common case */

    if (occupied < 0)
    {
        /* head has wrapped around, tail not yet */
        occupied += QUEUE_MAX_PAGE + 1;
    }

    return (double) occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2);
}

AsyncShmemInit()

void AsyncShmemInit

(

void

)

Definition at line 526 of file async.c.

References add_size(), asyncQueuePagePrecedes(), i, InvalidBackendId, InvalidOid, InvalidPid, AsyncQueueControl::lastQueueFillWarn, LWTRANCHE_NOTIFY_BUFFER, MaxBackends, mul_size(), NotifyCtl, NUM_NOTIFY_BUFFERS, offsetof, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_FIRST_LISTENER, QUEUE_HEAD, QUEUE_NEXT_LISTENER, QUEUE_STOP_PAGE, QUEUE_TAIL, SET_QUEUE_POS, ShmemInitStruct(), SimpleLruInit(), SlruScanDirCbDeleteAll(), SlruScanDirectory(), and SYNC_HANDLER_NONE.

Referenced by CreateSharedMemoryAndSemaphores().

{
    bool        found;
    Size        size;

    /*
     * Create or attach to the AsyncQueueControl structure.
     *
     * The used entries in the backend[] array run from 1 to MaxBackends; the
     * zero'th entry is unused but must be allocated.
     */
    size = mul_size(MaxBackends + 1, sizeof(QueueBackendStatus));
    size = add_size(size, offsetof(AsyncQueueControl, backend));

    asyncQueueControl = (AsyncQueueControl *)
        ShmemInitStruct("Async Queue Control", size, &found);

    if (!found)
    {
        /* First time through, so initialize it */
        SET_QUEUE_POS(QUEUE_HEAD, 0, 0);
        SET_QUEUE_POS(QUEUE_TAIL, 0, 0);
        QUEUE_STOP_PAGE = 0;
        QUEUE_FIRST_LISTENER = InvalidBackendId;
        asyncQueueControl->lastQueueFillWarn = 0;
        /* zero'th entry won't be used, but let's initialize it anyway */
        for (int i = 0; i <= MaxBackends; i++)
        {
            QUEUE_BACKEND_PID(i) = InvalidPid;
            QUEUE_BACKEND_DBOID(i) = InvalidOid;
            QUEUE_NEXT_LISTENER(i) = InvalidBackendId;
            SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0);
        }
    }

    /*
     * Set up SLRU management of the pg_notify data.
     */
    NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
    SimpleLruInit(NotifyCtl, "Notify", NUM_NOTIFY_BUFFERS, 0,
                  NotifySLRULock, "pg_notify", LWTRANCHE_NOTIFY_BUFFER,
                  SYNC_HANDLER_NONE);

    if (!found)
    {
        /*
         * During start or reboot, clean out the pg_notify directory.
         */
        (void) SlruScanDirectory(NotifyCtl, SlruScanDirCbDeleteAll, NULL);
    }
}

AsyncShmemSize()

Size AsyncShmemSize

(

void

)

Definition at line 509 of file async.c.

References add_size(), MaxBackends, mul_size(), NUM_NOTIFY_BUFFERS, offsetof, and SimpleLruShmemSize().

Referenced by CreateSharedMemoryAndSemaphores().

{
    Size        size;

    /* This had better match AsyncShmemInit */
    size = mul_size(MaxBackends + 1, sizeof(QueueBackendStatus));
    size = add_size(size, offsetof(AsyncQueueControl, backend));

    size = add_size(size, SimpleLruShmemSize(NUM_NOTIFY_BUFFERS, 0));

    return size;
}

AtAbort_Notify()

void AtAbort_Notify

(

void

)

Definition at line 1756 of file async.c.

References amRegisteredListener, asyncQueueUnregister(), ClearPendingActionsAndNotifies(), and NIL.

Referenced by AbortTransaction().

{
    /*
     * If we LISTEN but then roll back the transaction after PreCommit_Notify,
     * we have registered as a listener but have not made any entry in
     * listenChannels.  In that case, deregister again.
     */
    if (amRegisteredListener && listenChannels == NIL)
        asyncQueueUnregister();

    /* And clean up */
    ClearPendingActionsAndNotifies();
}

AtCommit_Notify()

void AtCommit_Notify

(

void

)

Definition at line 990 of file async.c.

References ListenAction::action, ActionList::actions, amRegisteredListener, asyncQueueUnregister(), ListenAction::channel, ClearPendingActionsAndNotifies(), DEBUG1, elog, Exec_ListenCommit(), Exec_UnlistenAllCommit(), Exec_UnlistenCommit(), lfirst, LISTEN_LISTEN, LISTEN_UNLISTEN, LISTEN_UNLISTEN_ALL, NIL, and Trace_notify.

Referenced by CommitTransaction().

{
    ListCell   *p;

    /*
     * Allow transactions that have not executed LISTEN/UNLISTEN/NOTIFY to
     * return as soon as possible
     */
    if (!pendingActions && !pendingNotifies)
        return;

    if (Trace_notify)
        elog(DEBUG1, "AtCommit_Notify");

    /* Perform any pending listen/unlisten actions */
    if (pendingActions != NULL)
    {
        foreach(p, pendingActions->actions)
        {
            ListenAction *actrec = (ListenAction *) lfirst(p);

            switch (actrec->action)
            {
                case LISTEN_LISTEN:
                    Exec_ListenCommit(actrec->channel);
                    break;
                case LISTEN_UNLISTEN:
                    Exec_UnlistenCommit(actrec->channel);
                    break;
                case LISTEN_UNLISTEN_ALL:
                    Exec_UnlistenAllCommit();
                    break;
            }
        }
    }

    /* If no longer listening to anything, get out of listener array */
    if (amRegisteredListener && listenChannels == NIL)
        asyncQueueUnregister();

    /* And clean up */
    ClearPendingActionsAndNotifies();
}

AtPrepare_Notify()

void AtPrepare_Notify

(

void

)

Definition at line 863 of file async.c.

References ereport, errcode(), errmsg(), and ERROR.

Referenced by PrepareTransaction().

{
    /* It's not allowed to have any pending LISTEN/UNLISTEN/NOTIFY actions */
    if (pendingActions || pendingNotifies)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot PREPARE a transaction that has executed LISTEN, UNLISTEN, or NOTIFY")));
}

AtSubAbort_Notify()

void AtSubAbort_Notify

(

void

)

Definition at line 1846 of file async.c.

References GetCurrentTransactionNestLevel(), ActionList::nestingLevel, NotificationList::nestingLevel, pendingActions, pendingNotifies, pfree(), ActionList::upper, and NotificationList::upper.

Referenced by AbortSubTransaction().

{
    int         my_level = GetCurrentTransactionNestLevel();

    /*
     * All we have to do is pop the stack --- the actions/notifies made in
     * this subxact are no longer interesting, and the space will be freed
     * when CurTransactionContext is recycled. We still have to free the
     * ActionList and NotificationList objects themselves, though, because
     * those are allocated in TopTransactionContext.
     *
     * Note that there might be no entries at all, or no entries for the
     * current subtransaction level, either because none were ever created, or
     * because we reentered this routine due to trouble during subxact abort.
     */
    while (pendingActions != NULL &&
           pendingActions->nestingLevel >= my_level)
    {
        ActionList *childPendingActions = pendingActions;

        pendingActions = pendingActions->upper;
        pfree(childPendingActions);
    }

    while (pendingNotifies != NULL &&
           pendingNotifies->nestingLevel >= my_level)
    {
        NotificationList *childPendingNotifies = pendingNotifies;

        pendingNotifies = pendingNotifies->upper;
        pfree(childPendingNotifies);
    }
}

AtSubCommit_Notify()

void AtSubCommit_Notify

(

void

)

Definition at line 1776 of file async.c.

References ActionList::actions, AddEventToPendingNotifies(), Assert, AsyncExistsPendingNotify(), NotificationList::events, GetCurrentTransactionNestLevel(), lfirst, list_concat(), ActionList::nestingLevel, NotificationList::nestingLevel, pendingActions, pendingNotifies, pfree(), ActionList::upper, and NotificationList::upper.

Referenced by CommitSubTransaction().

{
    int         my_level = GetCurrentTransactionNestLevel();

    /* If there are actions at our nesting level, we must reparent them. */
    if (pendingActions != NULL &&
        pendingActions->nestingLevel >= my_level)
    {
        if (pendingActions->upper == NULL ||
            pendingActions->upper->nestingLevel < my_level - 1)
        {
            /* nothing to merge; give the whole thing to the parent */
            --pendingActions->nestingLevel;
        }
        else
        {
            ActionList *childPendingActions = pendingActions;

            pendingActions = pendingActions->upper;

            /*
             * Mustn't try to eliminate duplicates here --- see queue_listen()
             */
            pendingActions->actions =
                list_concat(pendingActions->actions,
                            childPendingActions->actions);
            pfree(childPendingActions);
        }
    }

    /* If there are notifies at our nesting level, we must reparent them. */
    if (pendingNotifies != NULL &&
        pendingNotifies->nestingLevel >= my_level)
    {
        Assert(pendingNotifies->nestingLevel == my_level);

        if (pendingNotifies->upper == NULL ||
            pendingNotifies->upper->nestingLevel < my_level - 1)
        {
            /* nothing to merge; give the whole thing to the parent */
            --pendingNotifies->nestingLevel;
        }
        else
        {
            /*
             * Formerly, we didn't bother to eliminate duplicates here, but
             * now we must, else we fall foul of "Assert(!found)", either here
             * or during a later attempt to build the parent-level hashtable.
             */
            NotificationList *childPendingNotifies = pendingNotifies;
            ListCell   *l;

            pendingNotifies = pendingNotifies->upper;
            /* Insert all the subxact's events into parent, except for dups */
            foreach(l, childPendingNotifies->events)
            {
                Notification *childn = (Notification *) lfirst(l);

                if (!AsyncExistsPendingNotify(childn))
                    AddEventToPendingNotifies(childn);
            }
            pfree(childPendingNotifies);
        }
    }
}

ClearPendingActionsAndNotifies()

static void ClearPendingActionsAndNotifies ( void  )

static

Definition at line 2464 of file async.c.

Referenced by AtAbort_Notify(), and AtCommit_Notify().

{
    /*
     * Everything's allocated in either TopTransactionContext or the context
     * for the subtransaction to which it corresponds.  So, there's nothing to
     * do here except reset the pointers; the space will be reclaimed when the
     * contexts are deleted.
     */
    pendingActions = NULL;
    pendingNotifies = NULL;
}

Exec_ListenCommit()

static void Exec_ListenCommit ( const char *  channel )

static

Definition at line 1135 of file async.c.

References IsListeningOn(), lappend(), MemoryContextSwitchTo(), pstrdup(), and TopMemoryContext.

Referenced by AtCommit_Notify().

{
    MemoryContext oldcontext;

    /* Do nothing if we are already listening on this channel */
    if (IsListeningOn(channel))
        return;

    /*
     * Add the new channel name to listenChannels.
     *
     * XXX It is theoretically possible to get an out-of-memory failure here,
     * which would be bad because we already committed.  For the moment it
     * doesn't seem worth trying to guard against that, but maybe improve this
     * later.
     */
    oldcontext = MemoryContextSwitchTo(TopMemoryContext);
    listenChannels = lappend(listenChannels, pstrdup(channel));
    MemoryContextSwitchTo(oldcontext);
}

Exec_ListenPreCommit()

static void Exec_ListenPreCommit ( void  )

static

Definition at line 1040 of file async.c.

References amRegisteredListener, Async_UnlistenOnExit(), asyncQueueReadAllNotifications(), before_shmem_exit(), DEBUG1, elog, i, InvalidBackendId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyBackendId, MyDatabaseId, MyProcPid, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_FIRST_LISTENER, QUEUE_HEAD, QUEUE_NEXT_LISTENER, QUEUE_POS_EQUAL, QUEUE_POS_MAX, QUEUE_TAIL, Trace_notify, and unlistenExitRegistered.

Referenced by PreCommit_Notify().

{
    QueuePosition head;
    QueuePosition max;
    BackendId   prevListener;

    /*
     * Nothing to do if we are already listening to something, nor if we
     * already ran this routine in this transaction.
     */
    if (amRegisteredListener)
        return;

    if (Trace_notify)
        elog(DEBUG1, "Exec_ListenPreCommit(%d)", MyProcPid);

    /*
     * Before registering, make sure we will unlisten before dying. (Note:
     * this action does not get undone if we abort later.)
     */
    if (!unlistenExitRegistered)
    {
        before_shmem_exit(Async_UnlistenOnExit, 0);
        unlistenExitRegistered = true;
    }

    /*
     * This is our first LISTEN, so establish our pointer.
     *
     * We set our pointer to the global tail pointer and then move it forward
     * over already-committed notifications.  This ensures we cannot miss any
     * not-yet-committed notifications.  We might get a few more but that
     * doesn't hurt.
     *
     * In some scenarios there might be a lot of committed notifications that
     * have not yet been pruned away (because some backend is being lazy about
     * reading them).  To reduce our startup time, we can look at other
     * backends and adopt the maximum "pos" pointer of any backend that's in
     * our database; any notifications it's already advanced over are surely
     * committed and need not be re-examined by us.  (We must consider only
     * backends connected to our DB, because others will not have bothered to
     * check committed-ness of notifications in our DB.)
     *
     * We need exclusive lock here so we can look at other backends' entries
     * and manipulate the list links.
     */
    LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
    head = QUEUE_HEAD;
    max = QUEUE_TAIL;
    prevListener = InvalidBackendId;
    for (BackendId i = QUEUE_FIRST_LISTENER; i > 0; i = QUEUE_NEXT_LISTENER(i))
    {
        if (QUEUE_BACKEND_DBOID(i) == MyDatabaseId)
            max = QUEUE_POS_MAX(max, QUEUE_BACKEND_POS(i));
        /* Also find last listening backend before this one */
        if (i < MyBackendId)
            prevListener = i;
    }
    QUEUE_BACKEND_POS(MyBackendId) = max;
    QUEUE_BACKEND_PID(MyBackendId) = MyProcPid;
    QUEUE_BACKEND_DBOID(MyBackendId) = MyDatabaseId;
    /* Insert backend into list of listeners at correct position */
    if (prevListener > 0)
    {
        QUEUE_NEXT_LISTENER(MyBackendId) = QUEUE_NEXT_LISTENER(prevListener);
        QUEUE_NEXT_LISTENER(prevListener) = MyBackendId;
    }
    else
    {
        QUEUE_NEXT_LISTENER(MyBackendId) = QUEUE_FIRST_LISTENER;
        QUEUE_FIRST_LISTENER = MyBackendId;
    }
    LWLockRelease(NotifyQueueLock);

    /* Now we are listed in the global array, so remember we're listening */
    amRegisteredListener = true;

    /*
     * Try to move our pointer forward as far as possible.  This will skip
     * over already-committed notifications, which we want to do because they
     * might be quite stale.  Note that we are not yet listening on anything,
     * so we won't deliver such notifications to our frontend.  Also, although
     * our transaction might have executed NOTIFY, those message(s) aren't
     * queued yet so we won't skip them here.
     */
    if (!QUEUE_POS_EQUAL(max, head))
        asyncQueueReadAllNotifications();
}

Exec_UnlistenAllCommit()

static void Exec_UnlistenAllCommit ( void  )

static

Definition at line 1193 of file async.c.

References DEBUG1, elog, list_free_deep(), MyProcPid, NIL, and Trace_notify.

Referenced by Async_UnlistenOnExit(), and AtCommit_Notify().

{
    if (Trace_notify)
        elog(DEBUG1, "Exec_UnlistenAllCommit(%d)", MyProcPid);

    list_free_deep(listenChannels);
    listenChannels = NIL;
}

Exec_UnlistenCommit()

static void Exec_UnlistenCommit ( const char *  channel )

static

Definition at line 1162 of file async.c.

References DEBUG1, elog, foreach_delete_current, lfirst, MyProcPid, pfree(), and Trace_notify.

Referenced by AtCommit_Notify().

{
    ListCell   *q;

    if (Trace_notify)
        elog(DEBUG1, "Exec_UnlistenCommit(%s,%d)", channel, MyProcPid);

    foreach(q, listenChannels)
    {
        char       *lchan = (char *) lfirst(q);

        if (strcmp(lchan, channel) == 0)
        {
            listenChannels = foreach_delete_current(listenChannels, q);
            pfree(lchan);
            break;
        }
    }

    /*
     * We do not complain about unlistening something not being listened;
     * should we?
     */
}

HandleNotifyInterrupt()

void HandleNotifyInterrupt

(

void

)

Definition at line 1889 of file async.c.

References MyLatch, notifyInterruptPending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

{
    /*
     * Note: this is called by a SIGNAL HANDLER. You must be very wary what
     * you do here.
     */

    /* signal that work needs to be done */
    notifyInterruptPending = true;

    /* make sure the event is processed in due course */
    SetLatch(MyLatch);
}

IsListeningOn()

static bool IsListeningOn ( const char *  channel )

static

Definition at line 1290 of file async.c.

References lfirst.

Referenced by asyncQueueProcessPageEntries(), and Exec_ListenCommit().

{
    ListCell   *p;

    foreach(p, listenChannels)
    {
        char       *lchan = (char *) lfirst(p);

        if (strcmp(lchan, channel) == 0)
            return true;
    }
    return false;
}

notification_hash()

static uint32 notification_hash ( const void *  key, Size  keysize  )

static

Definition at line 2434 of file async.c.

References Assert, Notification::channel_len, Notification::data, DatumGetUInt32, hash_any(), and Notification::payload_len.

Referenced by AddEventToPendingNotifies().

{
    const Notification *k = *(const Notification *const *) key;

    Assert(keysize == sizeof(Notification *));
    /* We don't bother to include the payload's trailing null in the hash */
    return DatumGetUInt32(hash_any((const unsigned char *) k->data,
                                   k->channel_len + k->payload_len + 1));
}

notification_match()

static int notification_match ( const void *  key1, const void *  key2, Size  keysize  )

static

Definition at line 2448 of file async.c.

References Assert, Notification::channel_len, Notification::data, and Notification::payload_len.

Referenced by AddEventToPendingNotifies().

{
    const Notification *k1 = *(const Notification *const *) key1;
    const Notification *k2 = *(const Notification *const *) key2;

    Assert(keysize == sizeof(Notification *));
    if (k1->channel_len == k2->channel_len &&
        k1->payload_len == k2->payload_len &&
        memcmp(k1->data, k2->data,
               k1->channel_len + k1->payload_len + 2) == 0)
        return 0;               /* equal */
    return 1;                   /* not equal */
}

NotifyMyFrontEnd()

void NotifyMyFrontEnd	(	const char *	channel,
		const char *	payload,
		int32	srcPid
	)

Definition at line 2305 of file async.c.

References buf, DestRemote, elog, FrontendProtocol, INFO, PG_PROTOCOL_MAJOR, pq_beginmessage(), pq_endmessage(), pq_sendint32(), pq_sendstring(), and whereToSendOutput.

Referenced by asyncQueueProcessPageEntries(), and HandleParallelMessage().

{
    if (whereToSendOutput == DestRemote)
    {
        StringInfoData buf;

        pq_beginmessage(&buf, 'A');
        pq_sendint32(&buf, srcPid);
        pq_sendstring(&buf, channel);
        if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 3)
            pq_sendstring(&buf, payload);
        pq_endmessage(&buf);

        /*
         * NOTE: we do not do pq_flush() here.  For a self-notify, it will
         * happen at the end of the transaction, and for incoming notifies
         * ProcessIncomingNotify will do it after finding all the notifies.
         */
    }
    else
        elog(INFO, "NOTIFY for \"%s\" payload \"%s\"", channel, payload);
}

pg_listening_channels()

Datum pg_listening_channels

(

PG_FUNCTION_ARGS

)

Definition at line 817 of file async.c.

References FuncCallContext::call_cntr, CStringGetTextDatum, list_length(), list_nth(), SRF_FIRSTCALL_INIT, SRF_IS_FIRSTCALL, SRF_PERCALL_SETUP, SRF_RETURN_DONE, and SRF_RETURN_NEXT.

{
    FuncCallContext *funcctx;

    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL())
    {
        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();
    }

    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();

    if (funcctx->call_cntr < list_length(listenChannels))
    {
        char       *channel = (char *) list_nth(listenChannels,
                                                funcctx->call_cntr);

        SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(channel));
    }

    SRF_RETURN_DONE(funcctx);
}

pg_notification_queue_usage()

Datum pg_notification_queue_usage

(

PG_FUNCTION_ARGS

)

Definition at line 1566 of file async.c.

References asyncQueueAdvanceTail(), asyncQueueUsage(), LW_SHARED, LWLockAcquire(), LWLockRelease(), PG_RETURN_FLOAT8, and usage().

{
    double      usage;

    /* Advance the queue tail so we don't report a too-large result */
    asyncQueueAdvanceTail();

    LWLockAcquire(NotifyQueueLock, LW_SHARED);
    usage = asyncQueueUsage();
    LWLockRelease(NotifyQueueLock);

    PG_RETURN_FLOAT8(usage);
}

pg_notify()

Datum pg_notify

(

PG_FUNCTION_ARGS

)

Definition at line 584 of file async.c.

References Async_Notify(), PG_ARGISNULL, PG_GETARG_TEXT_PP, PG_RETURN_VOID, PreventCommandDuringRecovery(), and text_to_cstring().

{
    const char *channel;
    const char *payload;

    if (PG_ARGISNULL(0))
        channel = "";
    else
        channel = text_to_cstring(PG_GETARG_TEXT_PP(0));

    if (PG_ARGISNULL(1))
        payload = "";
    else
        payload = text_to_cstring(PG_GETARG_TEXT_PP(1));

    /* For NOTIFY as a statement, this is checked in ProcessUtility */
    PreventCommandDuringRecovery("NOTIFY");

    Async_Notify(channel, payload);

    PG_RETURN_VOID();
}

PreCommit_Notify()

void PreCommit_Notify

(

void

)

Definition at line 888 of file async.c.

References AccessExclusiveLock, ListenAction::action, ActionList::actions, asyncQueueAddEntries(), asyncQueueFillWarning(), asyncQueueIsFull(), backendHasSentNotifications, DEBUG1, elog, ereport, errcode(), errmsg(), ERROR, NotificationList::events, Exec_ListenPreCommit(), GetCurrentTransactionId(), InvalidOid, lfirst, list_head(), LISTEN_LISTEN, LISTEN_UNLISTEN, LISTEN_UNLISTEN_ALL, LockSharedObject(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), and Trace_notify.

Referenced by CommitTransaction().

{
    ListCell   *p;

    if (!pendingActions && !pendingNotifies)
        return;                 /* no relevant statements in this xact */

    if (Trace_notify)
        elog(DEBUG1, "PreCommit_Notify");

    /* Preflight for any pending listen/unlisten actions */
    if (pendingActions != NULL)
    {
        foreach(p, pendingActions->actions)
        {
            ListenAction *actrec = (ListenAction *) lfirst(p);

            switch (actrec->action)
            {
                case LISTEN_LISTEN:
                    Exec_ListenPreCommit();
                    break;
                case LISTEN_UNLISTEN:
                    /* there is no Exec_UnlistenPreCommit() */
                    break;
                case LISTEN_UNLISTEN_ALL:
                    /* there is no Exec_UnlistenAllPreCommit() */
                    break;
            }
        }
    }

    /* Queue any pending notifies (must happen after the above) */
    if (pendingNotifies)
    {
        ListCell   *nextNotify;

        /*
         * Make sure that we have an XID assigned to the current transaction.
         * GetCurrentTransactionId is cheap if we already have an XID, but not
         * so cheap if we don't, and we'd prefer not to do that work while
         * holding NotifyQueueLock.
         */
        (void) GetCurrentTransactionId();

        /*
         * Serialize writers by acquiring a special lock that we hold till
         * after commit.  This ensures that queue entries appear in commit
         * order, and in particular that there are never uncommitted queue
         * entries ahead of committed ones, so an uncommitted transaction
         * can't block delivery of deliverable notifications.
         *
         * We use a heavyweight lock so that it'll automatically be released
         * after either commit or abort.  This also allows deadlocks to be
         * detected, though really a deadlock shouldn't be possible here.
         *
         * The lock is on "database 0", which is pretty ugly but it doesn't
         * seem worth inventing a special locktag category just for this.
         * (Historical note: before PG 9.0, a similar lock on "database 0" was
         * used by the flatfiles mechanism.)
         */
        LockSharedObject(DatabaseRelationId, InvalidOid, 0,
                         AccessExclusiveLock);

        /* Now push the notifications into the queue */
        backendHasSentNotifications = true;

        nextNotify = list_head(pendingNotifies->events);
        while (nextNotify != NULL)
        {
            /*
             * Add the pending notifications to the queue.  We acquire and
             * release NotifyQueueLock once per page, which might be overkill
             * but it does allow readers to get in while we're doing this.
             *
             * A full queue is very uncommon and should really not happen,
             * given that we have so much space available in the SLRU pages.
             * Nevertheless we need to deal with this possibility. Note that
             * when we get here we are in the process of committing our
             * transaction, but we have not yet committed to clog, so at this
             * point in time we can still roll the transaction back.
             */
            LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
            asyncQueueFillWarning();
            if (asyncQueueIsFull())
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                         errmsg("too many notifications in the NOTIFY queue")));
            nextNotify = asyncQueueAddEntries(nextNotify);
            LWLockRelease(NotifyQueueLock);
        }
    }
}

ProcessCompletedNotifies()

void ProcessCompletedNotifies

(

void

)

Definition at line 1226 of file async.c.

References asyncQueueAdvanceTail(), asyncQueueReadAllNotifications(), backendHasSentNotifications, backendTryAdvanceTail, CommitTransactionCommand(), CurrentMemoryContext, DEBUG1, elog, MemoryContextSwitchTo(), NIL, SignalBackends(), StartTransactionCommand(), and Trace_notify.

Referenced by PostgresMain().

{
    MemoryContext caller_context;

    /* Nothing to do if we didn't send any notifications */
    if (!backendHasSentNotifications)
        return;

    /*
     * We reset the flag immediately; otherwise, if any sort of error occurs
     * below, we'd be locked up in an infinite loop, because control will come
     * right back here after error cleanup.
     */
    backendHasSentNotifications = false;

    /*
     * We must preserve the caller's memory context (probably MessageContext)
     * across the transaction we do here.
     */
    caller_context = CurrentMemoryContext;

    if (Trace_notify)
        elog(DEBUG1, "ProcessCompletedNotifies");

    /*
     * We must run asyncQueueReadAllNotifications inside a transaction, else
     * bad things happen if it gets an error.
     */
    StartTransactionCommand();

    /* Send signals to other backends */
    SignalBackends();

    if (listenChannels != NIL)
    {
        /* Read the queue ourselves, and send relevant stuff to the frontend */
        asyncQueueReadAllNotifications();
    }

    /*
     * If it's time to try to advance the global tail pointer, do that.
     */
    if (backendTryAdvanceTail)
    {
        backendTryAdvanceTail = false;
        asyncQueueAdvanceTail();
    }

    CommitTransactionCommand();

    MemoryContextSwitchTo(caller_context);

    /* We don't need pq_flush() here since postgres.c will do one shortly */
}

ProcessIncomingNotify()

static void ProcessIncomingNotify ( void  )

static

Definition at line 2266 of file async.c.

References asyncQueueReadAllNotifications(), CommitTransactionCommand(), DEBUG1, elog, NIL, notifyInterruptPending, pq_flush, set_ps_display(), StartTransactionCommand(), and Trace_notify.

Referenced by ProcessNotifyInterrupt().

{
    /* We *must* reset the flag */
    notifyInterruptPending = false;

    /* Do nothing else if we aren't actively listening */
    if (listenChannels == NIL)
        return;

    if (Trace_notify)
        elog(DEBUG1, "ProcessIncomingNotify");

    set_ps_display("notify interrupt");

    /*
     * We must run asyncQueueReadAllNotifications inside a transaction, else
     * bad things happen if it gets an error.
     */
    StartTransactionCommand();

    asyncQueueReadAllNotifications();

    CommitTransactionCommand();

    /*
     * Must flush the notify messages to ensure frontend gets them promptly.
     */
    pq_flush();

    set_ps_display("idle");

    if (Trace_notify)
        elog(DEBUG1, "ProcessIncomingNotify: done");
}

ProcessNotifyInterrupt()

void ProcessNotifyInterrupt

(

void

)

Definition at line 1915 of file async.c.

References IsTransactionOrTransactionBlock(), notifyInterruptPending, and ProcessIncomingNotify().

Referenced by PostgresMain(), and ProcessClientReadInterrupt().

{
    if (IsTransactionOrTransactionBlock())
        return;                 /* not really idle */

    while (notifyInterruptPending)
        ProcessIncomingNotify();
}

queue_listen()

static void queue_listen ( ListenActionKind  action, const char *  channel  )

static

Definition at line 717 of file async.c.

References generate_unaccent_rules::action, ListenAction::action, ActionList::actions, ListenAction::channel, CurTransactionContext, GetCurrentTransactionNestLevel(), lappend(), list_make1, MemoryContextAlloc(), MemoryContextSwitchTo(), ActionList::nestingLevel, offsetof, palloc(), pendingActions, TopTransactionContext, and ActionList::upper.

Referenced by Async_Listen(), Async_Unlisten(), and Async_UnlistenAll().

{
    MemoryContext oldcontext;
    ListenAction *actrec;
    int         my_level = GetCurrentTransactionNestLevel();

    /*
     * Unlike Async_Notify, we don't try to collapse out duplicates. It would
     * be too complicated to ensure we get the right interactions of
     * conflicting LISTEN/UNLISTEN/UNLISTEN_ALL, and it's unlikely that there
     * would be any performance benefit anyway in sane applications.
     */
    oldcontext = MemoryContextSwitchTo(CurTransactionContext);

    /* space for terminating null is included in sizeof(ListenAction) */
    actrec = (ListenAction *) palloc(offsetof(ListenAction, channel) +
                                     strlen(channel) + 1);
    actrec->action = action;
    strcpy(actrec->channel, channel);

    if (pendingActions == NULL || my_level > pendingActions->nestingLevel)
    {
        ActionList *actions;

        /*
         * First action in current sub(xact). Note that we allocate the
         * ActionList in TopTransactionContext; the nestingLevel might get
         * changed later by AtSubCommit_Notify.
         */
        actions = (ActionList *)
            MemoryContextAlloc(TopTransactionContext, sizeof(ActionList));
        actions->nestingLevel = my_level;
        actions->actions = list_make1(actrec);
        actions->upper = pendingActions;
        pendingActions = actions;
    }
    else
        pendingActions->actions = lappend(pendingActions->actions, actrec);

    MemoryContextSwitchTo(oldcontext);
}

SignalBackends()

static void SignalBackends ( void  )

static

Definition at line 1673 of file async.c.

References Assert, asyncQueuePageDiff(), DEBUG3, elog, i, InvalidPid, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MaxBackends, MyDatabaseId, MyProcPid, palloc(), pfree(), PROCSIG_NOTIFY_INTERRUPT, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_CLEANUP_DELAY, QUEUE_FIRST_LISTENER, QUEUE_HEAD, QUEUE_NEXT_LISTENER, QUEUE_POS_EQUAL, QUEUE_POS_PAGE, and SendProcSignal().

Referenced by ProcessCompletedNotifies().

{
    int32      *pids;
    BackendId  *ids;
    int         count;

    /*
     * Identify backends that we need to signal.  We don't want to send
     * signals while holding the NotifyQueueLock, so this loop just builds a
     * list of target PIDs.
     *
     * XXX in principle these pallocs could fail, which would be bad. Maybe
     * preallocate the arrays?  But in practice this is only run in trivial
     * transactions, so there should surely be space available.
     */
    pids = (int32 *) palloc(MaxBackends * sizeof(int32));
    ids = (BackendId *) palloc(MaxBackends * sizeof(BackendId));
    count = 0;

    LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
    for (BackendId i = QUEUE_FIRST_LISTENER; i > 0; i = QUEUE_NEXT_LISTENER(i))
    {
        int32       pid = QUEUE_BACKEND_PID(i);
        QueuePosition pos;

        Assert(pid != InvalidPid);
        if (pid == MyProcPid)
            continue;           /* never signal self */
        pos = QUEUE_BACKEND_POS(i);
        if (QUEUE_BACKEND_DBOID(i) == MyDatabaseId)
        {
            /*
             * Always signal listeners in our own database, unless they're
             * already caught up (unlikely, but possible).
             */
            if (QUEUE_POS_EQUAL(pos, QUEUE_HEAD))
                continue;
        }
        else
        {
            /*
             * Listeners in other databases should be signaled only if they
             * are far behind.
             */
            if (asyncQueuePageDiff(QUEUE_POS_PAGE(QUEUE_HEAD),
                                   QUEUE_POS_PAGE(pos)) < QUEUE_CLEANUP_DELAY)
                continue;
        }
        /* OK, need to signal this one */
        pids[count] = pid;
        ids[count] = i;
        count++;
    }
    LWLockRelease(NotifyQueueLock);

    /* Now send signals */
    for (int i = 0; i < count; i++)
    {
        int32       pid = pids[i];

        /*
         * Note: assuming things aren't broken, a signal failure here could
         * only occur if the target backend exited since we released
         * NotifyQueueLock; which is unlikely but certainly possible. So we
         * just log a low-level debug message if it happens.
         */
        if (SendProcSignal(pid, PROCSIG_NOTIFY_INTERRUPT, ids[i]) < 0)
            elog(DEBUG3, "could not signal backend with PID %d: %m", pid);
    }

    pfree(pids);
    pfree(ids);
}

Variable Documentation

amRegisteredListener

bool amRegisteredListener = false

static

Definition at line 427 of file async.c.

Referenced by asyncQueueUnregister(), AtAbort_Notify(), AtCommit_Notify(), and Exec_ListenPreCommit().

asyncQueueControl

AsyncQueueControl* asyncQueueControl

static

Definition at line 287 of file async.c.

backendHasSentNotifications

bool backendHasSentNotifications = false

static

Definition at line 430 of file async.c.

Referenced by PreCommit_Notify(), and ProcessCompletedNotifies().

backendTryAdvanceTail

bool backendTryAdvanceTail = false

static

Definition at line 433 of file async.c.

Referenced by asyncQueueAddEntries(), and ProcessCompletedNotifies().

listenChannels

List* listenChannels = NIL

static

Definition at line 328 of file async.c.

NotifyCtlData

SlruCtlData NotifyCtlData

static

Definition at line 301 of file async.c.

notifyInterruptPending

volatile sig_atomic_t notifyInterruptPending = false

Definition at line 421 of file async.c.

Referenced by HandleNotifyInterrupt(), PostgresMain(), ProcessClientReadInterrupt(), ProcessIncomingNotify(), and ProcessNotifyInterrupt().

pendingActions

ActionList* pendingActions = NULL

static

Definition at line 360 of file async.c.

Referenced by AtSubAbort_Notify(), AtSubCommit_Notify(), and queue_listen().

pendingNotifies

NotificationList* pendingNotifies = NULL

static

Definition at line 412 of file async.c.

Referenced by Async_Notify(), AtSubAbort_Notify(), and AtSubCommit_Notify().

Trace_notify

bool Trace_notify = false

Definition at line 436 of file async.c.

Referenced by Async_Listen(), Async_Notify(), Async_Unlisten(), Async_UnlistenAll(), AtCommit_Notify(), Exec_ListenPreCommit(), Exec_UnlistenAllCommit(), Exec_UnlistenCommit(), PreCommit_Notify(), ProcessCompletedNotifies(), and ProcessIncomingNotify().

unlistenExitRegistered

bool unlistenExitRegistered = false

static

Definition at line 424 of file async.c.

Referenced by Async_Unlisten(), Async_UnlistenAll(), and Exec_ListenPreCommit().