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/procsignal.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/guc_hooks.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	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

Enumerations
enum	ListenActionKind { LISTEN_LISTEN , LISTEN_UNLISTEN , LISTEN_UNLISTEN_ALL }

Functions
static int64	asyncQueuePageDiff (int64 p, int64 q)
static bool	asyncQueuePagePrecedes (int64 p, int64 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 (QueuePosition *current, QueuePosition stop, Snapshot snapshot)
static void	asyncQueueAdvanceTail (void)
static void	ProcessIncomingNotify (bool flush)
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)
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 (bool flush)
void	AsyncNotifyFreezeXids (TransactionId newFrozenXid)
void	NotifyMyFrontEnd (const char *channel, const char *payload, int32 srcPid)
bool	check_notify_buffers (int *newval, void **extra, GucSource source)

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	tryAdvanceTail = false
bool	Trace_notify = false
int	max_notify_queue_pages = 1048576

Macro Definition Documentation

AsyncQueueEntryEmptySize

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

Definition at line 189 of file async.c.

MIN_HASHABLE_NOTIFIES

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

Definition at line 397 of file async.c.

NOTIFY_PAYLOAD_MAX_LENGTH

#define NOTIFY_PAYLOAD_MAX_LENGTH   (BLCKSZ - NAMEDATALEN - 128)

Definition at line 163 of file async.c.

NotifyCtl

#define NotifyCtl   (&NotifyCtlData)

Definition at line 310 of file async.c.

QUEUE_BACKEND_DBOID

#define QUEUE_BACKEND_DBOID

(

i

)

(asyncQueueControl->backend[i].dboid)

Definition at line 301 of file async.c.

QUEUE_BACKEND_PID

#define QUEUE_BACKEND_PID

(

i

)

(asyncQueueControl->backend[i].pid)

Definition at line 300 of file async.c.

QUEUE_BACKEND_POS

#define QUEUE_BACKEND_POS

(

i

)

(asyncQueueControl->backend[i].pos)

Definition at line 303 of file async.c.

QUEUE_CLEANUP_DELAY

#define QUEUE_CLEANUP_DELAY   4

Definition at line 238 of file async.c.

QUEUE_FIRST_LISTENER

#define QUEUE_FIRST_LISTENER   (asyncQueueControl->firstListener)

Definition at line 299 of file async.c.

QUEUE_FULL_WARN_INTERVAL

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

Definition at line 313 of file async.c.

QUEUE_HEAD

#define QUEUE_HEAD   (asyncQueueControl->head)

Definition at line 296 of file async.c.

QUEUE_NEXT_LISTENER

#define QUEUE_NEXT_LISTENER

(

i

)

(asyncQueueControl->backend[i].nextListener)

Definition at line 302 of file async.c.

QUEUE_PAGESIZE

#define QUEUE_PAGESIZE   BLCKSZ

Definition at line 311 of file async.c.

QUEUE_POS_EQUAL

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

Definition at line 209 of file async.c.

QUEUE_POS_IS_ZERO

#define QUEUE_POS_IS_ZERO

(

x

)

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

Definition at line 212 of file async.c.

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 222 of file async.c.

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 216 of file async.c.

QUEUE_POS_OFFSET

#define QUEUE_POS_OFFSET

(

x

)

((x).offset)

Definition at line 201 of file async.c.

QUEUE_POS_PAGE

#define QUEUE_POS_PAGE

(

x

)

((x).page)

Definition at line 200 of file async.c.

QUEUE_STOP_PAGE

#define QUEUE_STOP_PAGE   (asyncQueueControl->stopPage)

Definition at line 298 of file async.c.

QUEUE_TAIL

#define QUEUE_TAIL   (asyncQueueControl->tail)

Definition at line 297 of file async.c.

QUEUEALIGN

#define QUEUEALIGN

(

len

)

INTALIGN(len)

Definition at line 187 of file async.c.

SET_QUEUE_POS

#define SET_QUEUE_POS	(		x,
			y,
			z
	)

Value:

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

Definition at line 203 of file async.c.

Typedef Documentation

ActionList

typedef struct ActionList ActionList

AsyncQueueControl

typedef struct AsyncQueueControl AsyncQueueControl

AsyncQueueEntry

typedef struct AsyncQueueEntry AsyncQueueEntry

Notification

typedef struct Notification Notification

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 332 of file async.c.

{
    LISTEN_LISTEN,
    LISTEN_UNLISTEN,
    LISTEN_UNLISTEN_ALL,
} ListenActionKind;

Function Documentation

AddEventToPendingNotifies()

static void AddEventToPendingNotifies ( Notification *  n )

static

Definition at line 2409 of file async.c.

{
    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(struct 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);
            bool        found;
 
            (void) hash_search(pendingNotifies->hashtab,
                               &oldn,
                               HASH_ENTER,
                               &found);
            Assert(!found);
        }
    }
 
    /* 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)
    {
        bool        found;
 
        (void) hash_search(pendingNotifies->hashtab,
                           &n,
                           HASH_ENTER,
                           &found);
        Assert(!found);
    }
}

References Assert(), CurTransactionContext, HASHCTL::entrysize, 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(), notification_match(), and pendingNotifies.

Referenced by Async_Notify(), and AtSubCommit_Notify().

Async_Listen()

void Async_Listen

(

const char *

channel

)

Definition at line 737 of file async.c.

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

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

Referenced by standard_ProcessUtility().

Async_Notify()

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

Definition at line 590 of file async.c.

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

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

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

Async_Unlisten()

void Async_Unlisten

(

const char *

channel

)

Definition at line 751 of file async.c.

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

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

Referenced by standard_ProcessUtility().

Async_UnlistenAll()

void Async_UnlistenAll

(

void

)

Definition at line 769 of file async.c.

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

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

Referenced by DiscardAll(), and standard_ProcessUtility().

Async_UnlistenOnExit()

static void Async_UnlistenOnExit ( int  code, Datum  arg  )

static

Definition at line 822 of file async.c.

{
    Exec_UnlistenAllCommit();
    asyncQueueUnregister();
}

References asyncQueueUnregister(), and Exec_UnlistenAllCommit().

Referenced by Exec_ListenPreCommit().

AsyncExistsPendingNotify()

static bool AsyncExistsPendingNotify ( Notification *  n )

static

Definition at line 2368 of file async.c.

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

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

Referenced by Async_Notify(), and AtSubCommit_Notify().

AsyncNotifyFreezeXids()

void AsyncNotifyFreezeXids

(

TransactionId

newFrozenXid

)

Definition at line 2196 of file async.c.

{
    QueuePosition pos;
    QueuePosition head;
    int64       curpage = -1;
    int         slotno = -1;
    char       *page_buffer = NULL;
    bool        page_dirty = false;
 
    /*
     * Acquire locks in the correct order to avoid deadlocks. As per the
     * locking protocol: NotifyQueueTailLock, then NotifyQueueLock, then SLRU
     * bank locks.
     *
     * We only need SHARED mode since we're just reading the head/tail
     * positions, not modifying them.
     */
    LWLockAcquire(NotifyQueueTailLock, LW_SHARED);
    LWLockAcquire(NotifyQueueLock, LW_SHARED);
 
    pos = QUEUE_TAIL;
    head = QUEUE_HEAD;
 
    /* Release NotifyQueueLock early, we only needed to read the positions */
    LWLockRelease(NotifyQueueLock);
 
    /*
     * Scan the queue from tail to head, freezing XIDs as needed. We hold
     * NotifyQueueTailLock throughout to ensure the tail doesn't move while
     * we're working.
     */
    while (!QUEUE_POS_EQUAL(pos, head))
    {
        AsyncQueueEntry *qe;
        TransactionId xid;
        int64       pageno = QUEUE_POS_PAGE(pos);
        int         offset = QUEUE_POS_OFFSET(pos);
 
        /* If we need a different page, release old lock and get new one */
        if (pageno != curpage)
        {
            LWLock     *lock;
 
            /* Release previous page if any */
            if (slotno >= 0)
            {
                if (page_dirty)
                {
                    NotifyCtl->shared->page_dirty[slotno] = true;
                    page_dirty = false;
                }
                LWLockRelease(SimpleLruGetBankLock(NotifyCtl, curpage));
            }
 
            lock = SimpleLruGetBankLock(NotifyCtl, pageno);
            LWLockAcquire(lock, LW_EXCLUSIVE);
            slotno = SimpleLruReadPage(NotifyCtl, pageno, true,
                                       InvalidTransactionId);
            page_buffer = NotifyCtl->shared->page_buffer[slotno];
            curpage = pageno;
        }
 
        qe = (AsyncQueueEntry *) (page_buffer + offset);
        xid = qe->xid;
 
        if (TransactionIdIsNormal(xid) &&
            TransactionIdPrecedes(xid, newFrozenXid))
        {
            if (TransactionIdDidCommit(xid))
            {
                qe->xid = FrozenTransactionId;
                page_dirty = true;
            }
            else
            {
                qe->xid = InvalidTransactionId;
                page_dirty = true;
            }
        }
 
        /* Advance to next entry */
        asyncQueueAdvance(&pos, qe->length);
    }
 
    /* Release final page lock if we acquired one */
    if (slotno >= 0)
    {
        if (page_dirty)
            NotifyCtl->shared->page_dirty[slotno] = true;
        LWLockRelease(SimpleLruGetBankLock(NotifyCtl, curpage));
    }
 
    LWLockRelease(NotifyQueueTailLock);
}

References asyncQueueAdvance(), FrozenTransactionId, InvalidTransactionId, AsyncQueueEntry::length, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), NotifyCtl, QUEUE_HEAD, QUEUE_POS_EQUAL, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, QUEUE_TAIL, SimpleLruGetBankLock(), SimpleLruReadPage(), TransactionIdDidCommit(), TransactionIdIsNormal, TransactionIdPrecedes(), and AsyncQueueEntry::xid.

Referenced by vac_truncate_clog().

asyncQueueAddEntries()

static ListCell * asyncQueueAddEntries ( ListCell *  nextNotify )

static

Definition at line 1355 of file async.c.

{
    AsyncQueueEntry qe;
    QueuePosition queue_head;
    int64       pageno;
    int         offset;
    int         slotno;
    LWLock     *prevlock;
 
    /*
     * 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.
     */
    pageno = QUEUE_POS_PAGE(queue_head);
    prevlock = SimpleLruGetBankLock(NotifyCtl, pageno);
 
    /* We hold both NotifyQueueLock and SLRU bank lock during this operation */
    LWLockAcquire(prevlock, LW_EXCLUSIVE);
 
    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.xid = InvalidTransactionId;
            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))
        {
            LWLock     *lock;
 
            pageno = QUEUE_POS_PAGE(queue_head);
            lock = SimpleLruGetBankLock(NotifyCtl, pageno);
            if (lock != prevlock)
            {
                LWLockRelease(prevlock);
                LWLockAcquire(lock, LW_EXCLUSIVE);
                prevlock = lock;
            }
 
            /*
             * 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)
                tryAdvanceTail = true;
 
            /* And exit the loop */
            break;
        }
    }
 
    /* Success, so update the global QUEUE_HEAD */
    QUEUE_HEAD = queue_head;
 
    LWLockRelease(prevlock);
 
    return nextNotify;
}

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

Referenced by PreCommit_Notify().

asyncQueueAdvance()

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

static

Definition at line 1286 of file async.c.

{
    int64       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++;
        offset = 0;
        pageJump = true;
    }
 
    SET_QUEUE_POS(*position, pageno, offset);
    return pageJump;
}

References Assert(), AsyncQueueEntryEmptySize, QUEUE_PAGESIZE, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, QUEUEALIGN, and SET_QUEUE_POS.

Referenced by AsyncNotifyFreezeXids(), asyncQueueAddEntries(), and asyncQueueProcessPageEntries().

asyncQueueAdvanceTail()

static void asyncQueueAdvanceTail ( void  )

static

Definition at line 2114 of file async.c.

{
    QueuePosition min;
    int64       oldtailpage;
    int64       newtailpage;
    int64       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 (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; 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 SLRU bank locks but will also
         * release the lock again.
         */
        SimpleLruTruncate(NotifyCtl, newtailpage);
 
        LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
        QUEUE_STOP_PAGE = newtailpage;
        LWLockRelease(NotifyQueueLock);
    }
 
    LWLockRelease(NotifyQueueTailLock);
}

References Assert(), asyncQueuePagePrecedes(), i, INVALID_PROC_NUMBER, 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 AtCommit_Notify(), and pg_notification_queue_usage().

asyncQueueFillWarning()

static void asyncQueueFillWarning ( void  )

static

Definition at line 1527 of file async.c.

{
    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 (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; 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;
    }
}

References Assert(), asyncQueueControl, asyncQueueUsage(), ereport, errdetail(), errhint(), errmsg(), GetCurrentTimestamp(), i, INVALID_PROC_NUMBER, 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().

asyncQueueIsFull()

static bool asyncQueueIsFull ( void  )

static

Definition at line 1271 of file async.c.

{
    int64       headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
    int64       tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
    int64       occupied = headPage - tailPage;
 
    return occupied >= max_notify_queue_pages;
}

References max_notify_queue_pages, QUEUE_HEAD, QUEUE_POS_PAGE, and QUEUE_TAIL.

Referenced by PreCommit_Notify().

asyncQueueNotificationToEntry()

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

static

Definition at line 1319 of file async.c.

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

References Assert(), AsyncQueueEntryEmptySize, Notification::channel_len, Notification::data, AsyncQueueEntry::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().

asyncQueuePageDiff()

static int64 asyncQueuePageDiff ( int64  p, int64  q  )

inlinestatic

Definition at line 465 of file async.c.

{
    return p - q;
}

Referenced by SignalBackends().

asyncQueuePagePrecedes()

static bool asyncQueuePagePrecedes ( int64  p, int64  q  )

inlinestatic

Definition at line 475 of file async.c.

{
    return p < q;
}

Referenced by asyncQueueAdvanceTail(), and AsyncShmemInit().

asyncQueueProcessPageEntries()

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

static

Definition at line 1977 of file async.c.

{
    int64       curpage = QUEUE_POS_PAGE(*current);
    int         slotno;
    char       *page_buffer;
    bool        reachedStop = false;
    bool        reachedEndOfPage;
 
    /*
     * We copy the entries into a local buffer to avoid holding the SLRU lock
     * while we transmit them to our frontend.  The local buffer must be
     * adequately aligned.
     */
    alignas(AsyncQueueEntry) char local_buf[QUEUE_PAGESIZE];
    char       *local_buf_end = local_buf;
 
    slotno = SimpleLruReadPage_ReadOnly(NotifyCtl, curpage,
                                        InvalidTransactionId);
    page_buffer = NotifyCtl->shared->page_buffer[slotno];
 
    do
    {
        QueuePosition thisentry = *current;
        AsyncQueueEntry *qe;
 
        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;
            }
 
            /*
             * Quick check for the case that we're not listening on any
             * channels, before calling TransactionIdDidCommit().  This makes
             * that case a little faster, but more importantly, it ensures
             * that if there's a bad entry in the queue for which
             * TransactionIdDidCommit() fails for some reason, we can skip
             * over it on the first LISTEN in a session, and not get stuck on
             * it indefinitely.
             */
            if (listenChannels == NIL)
                continue;
 
            if (TransactionIdDidCommit(qe->xid))
            {
                memcpy(local_buf_end, qe, qe->length);
                local_buf_end += qe->length;
            }
            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);
 
    /* Release lock that we got from SimpleLruReadPage_ReadOnly() */
    LWLockRelease(SimpleLruGetBankLock(NotifyCtl, curpage));
 
    /*
     * Now that we have let go of the SLRU bank lock, send the notifications
     * to our backend
     */
    Assert(local_buf_end - local_buf <= BLCKSZ);
    for (char *p = local_buf; p < local_buf_end;)
    {
        AsyncQueueEntry *qe = (AsyncQueueEntry *) p;
 
        /* 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);
        }
 
        p += qe->length;
    }
 
    if (QUEUE_POS_EQUAL(*current, stop))
        reachedStop = true;
 
    return reachedStop;
}

References Assert(), asyncQueueAdvance(), AsyncQueueEntry::data, AsyncQueueEntry::dboid, InvalidTransactionId, IsListeningOn(), AsyncQueueEntry::length, listenChannels, LWLockRelease(), MyDatabaseId, NIL, NotifyCtl, NotifyMyFrontEnd(), QUEUE_PAGESIZE, QUEUE_POS_EQUAL, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, SimpleLruGetBankLock(), SimpleLruReadPage_ReadOnly(), AsyncQueueEntry::srcPid, TransactionIdDidCommit(), AsyncQueueEntry::xid, and XidInMVCCSnapshot().

Referenced by asyncQueueReadAllNotifications().

asyncQueueReadAllNotifications()

static void asyncQueueReadAllNotifications ( void  )

static

Definition at line 1851 of file async.c.

{
    QueuePosition pos;
    QueuePosition head;
    Snapshot    snapshot;
 
    /* Fetch current state */
    LWLockAcquire(NotifyQueueLock, LW_SHARED);
    /* Assert checks that we have a valid state entry */
    Assert(MyProcPid == QUEUE_BACKEND_PID(MyProcNumber));
    pos = QUEUE_BACKEND_POS(MyProcNumber);
    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 set ExitOnAnyError to upgrade any
     * ERRORs to FATAL, causing the client connection to be closed on error.
     *
     * We used to only skip over the offending message and try to soldier on,
     * but it was somewhat questionable to lose a notification and give the
     * client an ERROR instead.  A client application is not be prepared for
     * that and can't tell that a notification was missed.  It was also not
     * very useful in practice because notifications are often processed while
     * a connection is idle and reading a message from the client, and in that
     * state, any error is upgraded to FATAL anyway.  Closing the connection
     * is a clear signal to the application that it might have missed
     * notifications.
     */
    {
        bool        save_ExitOnAnyError = ExitOnAnyError;
        bool        reachedStop;
 
        ExitOnAnyError = true;
 
        do
        {
            /*
             * 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, snapshot);
        } while (!reachedStop);
 
        /* Update shared state */
        LWLockAcquire(NotifyQueueLock, LW_SHARED);
        QUEUE_BACKEND_POS(MyProcNumber) = pos;
        LWLockRelease(NotifyQueueLock);
 
        ExitOnAnyError = save_ExitOnAnyError;
    }
 
    /* Done with snapshot */
    UnregisterSnapshot(snapshot);
}

References Assert(), asyncQueueProcessPageEntries(), ExitOnAnyError, GetLatestSnapshot(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyProcNumber, MyProcPid, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_HEAD, QUEUE_POS_EQUAL, RegisterSnapshot(), and UnregisterSnapshot().

Referenced by Exec_ListenPreCommit(), and ProcessIncomingNotify().

asyncQueueUnregister()

static void asyncQueueUnregister ( void  )

static

Definition at line 1230 of file async.c.

{
    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(MyProcNumber) = InvalidPid;
    QUEUE_BACKEND_DBOID(MyProcNumber) = InvalidOid;
    /* and remove it from the list */
    if (QUEUE_FIRST_LISTENER == MyProcNumber)
        QUEUE_FIRST_LISTENER = QUEUE_NEXT_LISTENER(MyProcNumber);
    else
    {
        for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i = QUEUE_NEXT_LISTENER(i))
        {
            if (QUEUE_NEXT_LISTENER(i) == MyProcNumber)
            {
                QUEUE_NEXT_LISTENER(i) = QUEUE_NEXT_LISTENER(MyProcNumber);
                break;
            }
        }
    }
    QUEUE_NEXT_LISTENER(MyProcNumber) = INVALID_PROC_NUMBER;
    LWLockRelease(NotifyQueueLock);
 
    /* mark ourselves as no longer listed in the global array */
    amRegisteredListener = false;
}

References amRegisteredListener, Assert(), i, INVALID_PROC_NUMBER, InvalidOid, InvalidPid, listenChannels, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, NIL, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_FIRST_LISTENER, and QUEUE_NEXT_LISTENER.

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

asyncQueueUsage()

static double asyncQueueUsage ( void  )

static

Definition at line 1506 of file async.c.

{
    int64       headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
    int64       tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
    int64       occupied = headPage - tailPage;
 
    if (occupied == 0)
        return (double) 0;      /* fast exit for common case */
 
    return (double) occupied / (double) max_notify_queue_pages;
}

References max_notify_queue_pages, QUEUE_HEAD, QUEUE_POS_PAGE, and QUEUE_TAIL.

Referenced by asyncQueueFillWarning(), and pg_notification_queue_usage().

AsyncShmemInit()

void AsyncShmemInit

(

void

)

Definition at line 501 of file async.c.

{
    bool        found;
    Size        size;
 
    /*
     * Create or attach to the AsyncQueueControl structure.
     */
    size = mul_size(MaxBackends, 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 = INVALID_PROC_NUMBER;
        asyncQueueControl->lastQueueFillWarn = 0;
        for (int i = 0; i < MaxBackends; i++)
        {
            QUEUE_BACKEND_PID(i) = InvalidPid;
            QUEUE_BACKEND_DBOID(i) = InvalidOid;
            QUEUE_NEXT_LISTENER(i) = INVALID_PROC_NUMBER;
            SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0);
        }
    }
 
    /*
     * Set up SLRU management of the pg_notify data. Note that long segment
     * names are used in order to avoid wraparound.
     */
    NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
    SimpleLruInit(NotifyCtl, "notify", notify_buffers, 0,
                  "pg_notify", LWTRANCHE_NOTIFY_BUFFER, LWTRANCHE_NOTIFY_SLRU,
                  SYNC_HANDLER_NONE, true);
 
    if (!found)
    {
        /*
         * During start or reboot, clean out the pg_notify directory.
         */
        (void) SlruScanDirectory(NotifyCtl, SlruScanDirCbDeleteAll, NULL);
    }
}

References add_size(), asyncQueueControl, asyncQueuePagePrecedes(), i, INVALID_PROC_NUMBER, InvalidOid, InvalidPid, AsyncQueueControl::lastQueueFillWarn, MaxBackends, mul_size(), notify_buffers, NotifyCtl, 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 CreateOrAttachShmemStructs().

AsyncShmemSize()

Size AsyncShmemSize

(

void

)

Definition at line 484 of file async.c.

{
    Size        size;
 
    /* This had better match AsyncShmemInit */
    size = mul_size(MaxBackends, sizeof(QueueBackendStatus));
    size = add_size(size, offsetof(AsyncQueueControl, backend));
 
    size = add_size(size, SimpleLruShmemSize(notify_buffers, 0));
 
    return size;
}

References add_size(), MaxBackends, mul_size(), notify_buffers, and SimpleLruShmemSize().

Referenced by CalculateShmemSize().

AtAbort_Notify()

void AtAbort_Notify

(

void

)

Definition at line 1671 of file async.c.

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

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

Referenced by AbortTransaction().

AtCommit_Notify()

void AtCommit_Notify

(

void

)

Definition at line 967 of file async.c.

{
    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();
 
    /*
     * Send signals to listening backends.  We need do this only if there are
     * pending notifies, which were previously added to the shared queue by
     * PreCommit_Notify().
     */
    if (pendingNotifies != NULL)
        SignalBackends();
 
    /*
     * If it's time to try to advance the global tail pointer, do that.
     *
     * (It might seem odd to do this in the sender, when more than likely the
     * listeners won't yet have read the messages we just sent.  However,
     * there's less contention if only the sender does it, and there is little
     * need for urgency in advancing the global tail.  So this typically will
     * be clearing out messages that were sent some time ago.)
     */
    if (tryAdvanceTail)
    {
        tryAdvanceTail = false;
        asyncQueueAdvanceTail();
    }
 
    /* And clean up */
    ClearPendingActionsAndNotifies();
}

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

Referenced by CommitTransaction().

AtPrepare_Notify()

void AtPrepare_Notify

(

void

)

Definition at line 835 of file async.c.

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

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

Referenced by PrepareTransaction().

AtSubAbort_Notify()

void AtSubAbort_Notify

(

void

)

Definition at line 1761 of file async.c.

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

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

Referenced by AbortSubTransaction().

AtSubCommit_Notify()

void AtSubCommit_Notify

(

void

)

Definition at line 1691 of file async.c.

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

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().

check_notify_buffers()

bool check_notify_buffers	(	int *	newval,
		void **	extra,
		GucSource	source
	)

Definition at line 2514 of file async.c.

{
    return check_slru_buffers("notify_buffers", newval);
}

References check_slru_buffers(), and newval.

ClearPendingActionsAndNotifies()

static void ClearPendingActionsAndNotifies ( void  )

static

Definition at line 2498 of file async.c.

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

References pendingActions, and pendingNotifies.

Referenced by AtAbort_Notify(), and AtCommit_Notify().

Exec_ListenCommit()

static void Exec_ListenCommit ( const char *  channel )

static

Definition at line 1135 of file async.c.

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

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

Referenced by AtCommit_Notify().

Exec_ListenPreCommit()

static void Exec_ListenPreCommit ( void  )

static

Definition at line 1040 of file async.c.

{
    QueuePosition head;
    QueuePosition max;
    ProcNumber  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 = INVALID_PROC_NUMBER;
    for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; 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 < MyProcNumber)
            prevListener = i;
    }
    QUEUE_BACKEND_POS(MyProcNumber) = max;
    QUEUE_BACKEND_PID(MyProcNumber) = MyProcPid;
    QUEUE_BACKEND_DBOID(MyProcNumber) = MyDatabaseId;
    /* Insert backend into list of listeners at correct position */
    if (prevListener != INVALID_PROC_NUMBER)
    {
        QUEUE_NEXT_LISTENER(MyProcNumber) = QUEUE_NEXT_LISTENER(prevListener);
        QUEUE_NEXT_LISTENER(prevListener) = MyProcNumber;
    }
    else
    {
        QUEUE_NEXT_LISTENER(MyProcNumber) = QUEUE_FIRST_LISTENER;
        QUEUE_FIRST_LISTENER = MyProcNumber;
    }
    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();
}

References amRegisteredListener, Async_UnlistenOnExit(), asyncQueueReadAllNotifications(), before_shmem_exit(), DEBUG1, elog, i, INVALID_PROC_NUMBER, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyDatabaseId, MyProcNumber, 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().

Exec_UnlistenAllCommit()

static void Exec_UnlistenAllCommit ( void  )

static

Definition at line 1193 of file async.c.

{
    if (Trace_notify)
        elog(DEBUG1, "Exec_UnlistenAllCommit(%d)", MyProcPid);
 
    list_free_deep(listenChannels);
    listenChannels = NIL;
}

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

Referenced by Async_UnlistenOnExit(), and AtCommit_Notify().

Exec_UnlistenCommit()

static void Exec_UnlistenCommit ( const char *  channel )

static

Definition at line 1162 of file async.c.

{
    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?
     */
}

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

Referenced by AtCommit_Notify().

HandleNotifyInterrupt()

void HandleNotifyInterrupt

(

void

)

Definition at line 1804 of file async.c.

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

References MyLatch, notifyInterruptPending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

IsListeningOn()

static bool IsListeningOn ( const char *  channel )

static

Definition at line 1211 of file async.c.

{
    ListCell   *p;
 
    foreach(p, listenChannels)
    {
        char       *lchan = (char *) lfirst(p);
 
        if (strcmp(lchan, channel) == 0)
            return true;
    }
    return false;
}

References lfirst, and listenChannels.

Referenced by asyncQueueProcessPageEntries(), and Exec_ListenCommit().

notification_hash()

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

static

Definition at line 2468 of file async.c.

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

References Assert(), Notification::channel_len, Notification::data, DatumGetUInt32(), hash_any(), sort-test::key, and Notification::payload_len.

Referenced by AddEventToPendingNotifies().

notification_match()

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

static

Definition at line 2482 of file async.c.

{
    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 */
}

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

Referenced by AddEventToPendingNotifies().

NotifyMyFrontEnd()

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

Definition at line 2344 of file async.c.

{
    if (whereToSendOutput == DestRemote)
    {
        StringInfoData buf;
 
        pq_beginmessage(&buf, PqMsg_NotificationResponse);
        pq_sendint32(&buf, srcPid);
        pq_sendstring(&buf, channel);
        pq_sendstring(&buf, payload);
        pq_endmessage(&buf);
 
        /*
         * NOTE: we do not do pq_flush() here.  Some level of caller will
         * handle it later, allowing this message to be combined into a packet
         * with other ones.
         */
    }
    else
        elog(INFO, "NOTIFY for \"%s\" payload \"%s\"", channel, payload);
}

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

Referenced by asyncQueueProcessPageEntries(), and ProcessParallelMessage().

pg_listening_channels()

Datum pg_listening_channels

(

PG_FUNCTION_ARGS

)

Definition at line 789 of file async.c.

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

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

pg_notification_queue_usage()

Datum pg_notification_queue_usage

(

PG_FUNCTION_ARGS

)

Definition at line 1481 of file async.c.

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

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

pg_notify()

Datum pg_notify

(

PG_FUNCTION_ARGS

)

Definition at line 556 of file async.c.

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

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

PreCommit_Notify()

void PreCommit_Notify

(

void

)

Definition at line 860 of file async.c.

{
    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 */
        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);
        }
 
        /* Note that we don't clear pendingNotifies; AtCommit_Notify will. */
    }
}

References AccessExclusiveLock, ListenAction::action, ActionList::actions, asyncQueueAddEntries(), asyncQueueFillWarning(), asyncQueueIsFull(), 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(), pendingActions, pendingNotifies, and Trace_notify.

Referenced by CommitTransaction().

ProcessIncomingNotify()

static void ProcessIncomingNotify ( bool  flush )

static

Definition at line 2303 of file async.c.

{
    /* 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();
 
    /*
     * If this isn't an end-of-command case, we must flush the notify messages
     * to ensure frontend gets them promptly.
     */
    if (flush)
        pq_flush();
 
    set_ps_display("idle");
 
    if (Trace_notify)
        elog(DEBUG1, "ProcessIncomingNotify: done");
}

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

Referenced by ProcessNotifyInterrupt().

ProcessNotifyInterrupt()

void ProcessNotifyInterrupt

(

bool

flush

)

Definition at line 1834 of file async.c.

{
    if (IsTransactionOrTransactionBlock())
        return;                 /* not really idle */
 
    /* Loop in case another signal arrives while sending messages */
    while (notifyInterruptPending)
        ProcessIncomingNotify(flush);
}

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

Referenced by PostgresMain(), and ProcessClientReadInterrupt().

queue_listen()

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

static

Definition at line 689 of file async.c.

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

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

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

SignalBackends()

static void SignalBackends ( void  )

static

Definition at line 1581 of file async.c.

{
    int32      *pids;
    ProcNumber *procnos;
    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?  They're not that large, though.
     */
    pids = (int32 *) palloc(MaxBackends * sizeof(int32));
    procnos = (ProcNumber *) palloc(MaxBackends * sizeof(ProcNumber));
    count = 0;
 
    LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
    for (ProcNumber i = QUEUE_FIRST_LISTENER; i != INVALID_PROC_NUMBER; i = QUEUE_NEXT_LISTENER(i))
    {
        int32       pid = QUEUE_BACKEND_PID(i);
        QueuePosition pos;
 
        Assert(pid != InvalidPid);
        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;
        procnos[count] = i;
        count++;
    }
    LWLockRelease(NotifyQueueLock);
 
    /* Now send signals */
    for (int i = 0; i < count; i++)
    {
        int32       pid = pids[i];
 
        /*
         * If we are signaling our own process, no need to involve the kernel;
         * just set the flag directly.
         */
        if (pid == MyProcPid)
        {
            notifyInterruptPending = true;
            continue;
        }
 
        /*
         * 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, procnos[i]) < 0)
            elog(DEBUG3, "could not signal backend with PID %d: %m", pid);
    }
 
    pfree(pids);
    pfree(procnos);
}

References Assert(), asyncQueuePageDiff(), DEBUG3, elog, i, INVALID_PROC_NUMBER, InvalidPid, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MaxBackends, MyDatabaseId, MyProcPid, notifyInterruptPending, 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 AtCommit_Notify().

Variable Documentation

amRegisteredListener

bool amRegisteredListener = false

static

Definition at line 419 of file async.c.

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

asyncQueueControl

AsyncQueueControl* asyncQueueControl

static

Definition at line 294 of file async.c.

Referenced by asyncQueueFillWarning(), and AsyncShmemInit().

listenChannels

List* listenChannels = NIL

static

Definition at line 320 of file async.c.

Referenced by asyncQueueProcessPageEntries(), asyncQueueUnregister(), AtAbort_Notify(), AtCommit_Notify(), Exec_ListenCommit(), Exec_UnlistenAllCommit(), Exec_UnlistenCommit(), IsListeningOn(), pg_listening_channels(), and ProcessIncomingNotify().

max_notify_queue_pages

int max_notify_queue_pages = 1048576

Definition at line 428 of file async.c.

Referenced by asyncQueueIsFull(), and asyncQueueUsage().

NotifyCtlData

SlruCtlData NotifyCtlData

static

Definition at line 308 of file async.c.

notifyInterruptPending

volatile sig_atomic_t notifyInterruptPending = false

Definition at line 413 of file async.c.

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

pendingActions

ActionList* pendingActions = NULL

static

Definition at line 352 of file async.c.

Referenced by Async_Unlisten(), Async_UnlistenAll(), AtCommit_Notify(), AtPrepare_Notify(), AtSubAbort_Notify(), AtSubCommit_Notify(), ClearPendingActionsAndNotifies(), PreCommit_Notify(), and queue_listen().

pendingNotifies

NotificationList* pendingNotifies = NULL

static

Definition at line 404 of file async.c.

Referenced by AddEventToPendingNotifies(), Async_Notify(), AsyncExistsPendingNotify(), asyncQueueAddEntries(), AtCommit_Notify(), AtPrepare_Notify(), AtSubAbort_Notify(), AtSubCommit_Notify(), ClearPendingActionsAndNotifies(), and PreCommit_Notify().

Trace_notify

bool Trace_notify = false

Definition at line 425 of file async.c.

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

tryAdvanceTail

bool tryAdvanceTail = false

static

Definition at line 422 of file async.c.

Referenced by asyncQueueAddEntries(), and AtCommit_Notify().

unlistenExitRegistered

bool unlistenExitRegistered = false

static

Definition at line 416 of file async.c.

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