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 "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/hashutils.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	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_MIN(x, y)
#define	QUEUE_POS_MAX(x, y)
#define	QUEUE_HEAD   (asyncQueueControl->head)
#define	QUEUE_TAIL   (asyncQueueControl->tail)
#define	QUEUE_BACKEND_PID(i)   (asyncQueueControl->backend[i].pid)
#define	QUEUE_BACKEND_DBOID(i)   (asyncQueueControl->backend[i].dboid)
#define	QUEUE_BACKEND_POS(i)   (asyncQueueControl->backend[i].pos)
#define	AsyncCtl   (&AsyncCtlData)
#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 Notification	Notification
typedef struct NotificationList	NotificationList
typedef struct NotificationHash	NotificationHash

Enumerations
enum	ListenActionKind { LISTEN_LISTEN, LISTEN_UNLISTEN, LISTEN_UNLISTEN_ALL }

Functions
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 bool	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	AtSubStart_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	AsyncCtlData
static List *	listenChannels = NIL
static List *	pendingActions = NIL
static List *	upperPendingActions = NIL
static NotificationList *	pendingNotifies = NULL
static List *	upperPendingNotifies = NIL
volatile sig_atomic_t	notifyInterruptPending = false
static bool	unlistenExitRegistered = false
static bool	amRegisteredListener = false
static bool	backendHasSentNotifications = false
bool	Trace_notify = false

Macro Definition Documentation

AsyncCtl

#define AsyncCtl   (&AsyncCtlData)

Definition at line 268 of file async.c.

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

AsyncQueueEntryEmptySize

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

Definition at line 179 of file async.c.

Referenced by asyncQueueAdvance(), and asyncQueueNotificationToEntry().

MIN_HASHABLE_NOTIFIES

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

Definition at line 366 of file async.c.

Referenced by AddEventToPendingNotifies().

NOTIFY_PAYLOAD_MAX_LENGTH

#define NOTIFY_PAYLOAD_MAX_LENGTH   (BLCKSZ - NAMEDATALEN - 128)

Definition at line 153 of file async.c.

Referenced by Async_Notify(), and asyncQueueNotificationToEntry().

QUEUE_BACKEND_DBOID

#define QUEUE_BACKEND_DBOID

(

i

)

(asyncQueueControl->backend[i].dboid)

Definition at line 260 of file async.c.

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

QUEUE_BACKEND_PID

#define QUEUE_BACKEND_PID

(

i

)

(asyncQueueControl->backend[i].pid)

Definition at line 259 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 261 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueFillWarning(), asyncQueueReadAllNotifications(), 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 270 of file async.c.

Referenced by asyncQueueFillWarning().

QUEUE_HEAD

#define QUEUE_HEAD   (asyncQueueControl->head)

Definition at line 257 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 289 of file async.c.

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

QUEUE_PAGESIZE

#define QUEUE_PAGESIZE   BLCKSZ

Definition at line 269 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 199 of file async.c.

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

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

Referenced by asyncQueueAdvanceTail(), and asyncQueueFillWarning().

QUEUE_POS_OFFSET

#define QUEUE_POS_OFFSET

(

x

)

((x).offset)

Definition at line 191 of file async.c.

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

QUEUE_POS_PAGE

#define QUEUE_POS_PAGE

(

x

)

((x).page)

Definition at line 190 of file async.c.

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

QUEUE_TAIL

#define QUEUE_TAIL   (asyncQueueControl->tail)

Definition at line 258 of file async.c.

Referenced by asyncQueueAdvanceTail(), asyncQueueIsFull(), asyncQueueReadAllNotifications(), asyncQueueUnregister(), asyncQueueUsage(), AsyncShmemInit(), and Exec_ListenPreCommit().

QUEUEALIGN

#define QUEUEALIGN

(

len

)

INTALIGN(len)

Definition at line 177 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 193 of file async.c.

Referenced by asyncQueueAdvance(), and AsyncShmemInit().

Typedef Documentation

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

{
    LISTEN_LISTEN,
    LISTEN_UNLISTEN,
    LISTEN_UNLISTEN_ALL
} ListenActionKind;

Function Documentation

AddEventToPendingNotifies()

static void AddEventToPendingNotifies ( Notification *  n )

static

Definition at line 2220 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, MemSet, 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 */
        MemSet(&hash_ctl, 0, sizeof(hash_ctl));
        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 689 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 572 of file async.c.

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

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

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

    /* Now check for duplicates */
    if (AsyncExistsPendingNotify(n))
    {
        /* It's a dup, so forget it */
        pfree(n);
        MemoryContextSwitchTo(oldcontext);
        return;
    }

    if (pendingNotifies == NULL)
    {
        /* First notify event in current (sub)xact */
        pendingNotifies = (NotificationList *) palloc(sizeof(NotificationList));
        pendingNotifies->events = list_make1(n);
        /* We certainly don't need a hashtable yet */
        pendingNotifies->hashtab = NULL;
    }
    else
    {
        /* Append more events to existing list */
        AddEventToPendingNotifies(n);
    }

    MemoryContextSwitchTo(oldcontext);
}

Async_Unlisten()

void Async_Unlisten

(

const char *

channel

)

Definition at line 703 of file async.c.

References DEBUG1, elog, LISTEN_UNLISTEN, MyProcPid, NIL, 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 == NIL && !unlistenExitRegistered)
        return;

    queue_listen(LISTEN_UNLISTEN, channel);
}

Async_UnlistenAll()

void Async_UnlistenAll

(

void

)

Definition at line 721 of file async.c.

References DEBUG1, elog, LISTEN_UNLISTEN_ALL, MyProcPid, NIL, 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 == NIL && !unlistenExitRegistered)
        return;

    queue_listen(LISTEN_UNLISTEN_ALL, "");
}

Async_UnlistenOnExit()

static void Async_UnlistenOnExit ( int  code, Datum  arg  )

static

Definition at line 774 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 2179 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 1356 of file async.c.

References AsyncCtl, asyncQueueAdvance(), asyncQueueNotificationToEntry(), AsyncQueueEntry::data, AsyncQueueEntry::dboid, NotificationList::events, InvalidOid, InvalidTransactionId, AsyncQueueEntry::length, lfirst, lnext(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), QUEUE_HEAD, QUEUE_PAGESIZE, 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 AsyncQueueLock and AsyncCtlLock during this operation */
    LWLockAcquire(AsyncCtlLock, 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;

    /* Fetch the current page */
    pageno = QUEUE_POS_PAGE(queue_head);
    slotno = SimpleLruReadPage(AsyncCtl, pageno, true, InvalidTransactionId);
    /* Note we mark the page dirty before writing in it */
    AsyncCtl->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(AsyncCtl->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(AsyncCtl, QUEUE_POS_PAGE(queue_head));
            /* And exit the loop */
            break;
        }
    }

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

    LWLockRelease(AsyncCtlLock);

    return nextNotify;
}

asyncQueueAdvance()

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

static

Definition at line 1285 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 2064 of file async.c.

References AsyncCtl, asyncQueuePagePrecedes(), i, InvalidPid, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MaxBackends, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_HEAD, QUEUE_POS_MIN, QUEUE_POS_PAGE, QUEUE_TAIL, SimpleLruTruncate(), and SLRU_PAGES_PER_SEGMENT.

Referenced by asyncQueueReadAllNotifications(), asyncQueueUnregister(), and ProcessCompletedNotifies().

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

    LWLockAcquire(AsyncQueueLock, LW_EXCLUSIVE);
    min = QUEUE_HEAD;
    for (i = 1; i <= MaxBackends; i++)
    {
        if (QUEUE_BACKEND_PID(i) != InvalidPid)
            min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i));
    }
    oldtailpage = QUEUE_POS_PAGE(QUEUE_TAIL);
    QUEUE_TAIL = min;
    LWLockRelease(AsyncQueueLock);

    /*
     * 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 AsyncCtlLock but will also release
         * the lock again.
         */
        SimpleLruTruncate(AsyncCtl, newtailpage);
    }
}

asyncQueueFillWarning()

static void asyncQueueFillWarning ( void  )

static

Definition at line 1495 of file async.c.

References asyncQueueUsage(), ereport, errdetail(), errhint(), errmsg(), GetCurrentTimestamp(), i, InvalidPid, AsyncQueueControl::lastQueueFillWarn, MaxBackends, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_FULL_WARN_INTERVAL, QUEUE_HEAD, 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;
        int         i;

        for (i = 1; i <= MaxBackends; i++)
        {
            if (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 1254 of file async.c.

References asyncQueuePagePrecedes(), QUEUE_HEAD, QUEUE_MAX_PAGE, QUEUE_POS_PAGE, QUEUE_TAIL, 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 (compare the truncation logic in
     * asyncQueueAdvanceTail).
     *
     * 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_POS_PAGE(QUEUE_TAIL);
    boundary -= boundary % SLRU_PAGES_PER_SEGMENT;
    return asyncQueuePagePrecedes(nexthead, boundary);
}

asyncQueueNotificationToEntry()

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

static

Definition at line 1320 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);
}

asyncQueuePagePrecedes()

static bool asyncQueuePagePrecedes ( int  p, int  q  )

static

Definition at line 432 of file async.c.

References Assert, and QUEUE_MAX_PAGE.

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

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

asyncQueueProcessPageEntries()

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

static

Definition at line 1975 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 1805 of file async.c.

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

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

{
    volatile QueuePosition pos;
    QueuePosition oldpos;
    QueuePosition head;
    Snapshot    snapshot;
    bool        advanceTail;

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

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

    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 */
    snapshot = RegisterSnapshot(GetLatestSnapshot());

    /*----------
     * Note that we deliver everything that we see in the queue and that
     * matches our _current_ listening state.
     * Especially we do not take into account different commit times.
     * Consider the following example:
     *
     * Backend 1:                    Backend 2:
     *
     * transaction starts
     * NOTIFY foo;
     * commit starts
     *                               transaction starts
     *                               LISTEN foo;
     *                               commit starts
     * commit to clog
     *                               commit to clog
     *
     * It could happen that backend 2 sees the notification from backend 1 in
     * the queue.  Even though the notifying transaction committed before
     * the listening transaction, we still deliver the notification.
     *
     * The idea is that an additional notification does not do any harm, we
     * just need to make sure that we do not miss a notification.
     *
     * 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 backend position before we lose
     * control to an error.  (We could alternatively retake AsyncQueueLock
     * 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 AsyncCtlLock 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(AsyncCtl, 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,
                   AsyncCtl->shared->page_buffer[slotno] + curoffset,
                   copysize);
            /* Release lock that we got from SimpleLruReadPage_ReadOnly() */
            LWLockRelease(AsyncCtlLock);

            /*
             * 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 AsyncQueueLock 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_CATCH();
    {
        /* Update shared state */
        LWLockAcquire(AsyncQueueLock, LW_SHARED);
        QUEUE_BACKEND_POS(MyBackendId) = pos;
        advanceTail = QUEUE_POS_EQUAL(oldpos, QUEUE_TAIL);
        LWLockRelease(AsyncQueueLock);

        /* If we were the laziest backend, try to advance the tail pointer */
        if (advanceTail)
            asyncQueueAdvanceTail();

        PG_RE_THROW();
    }
    PG_END_TRY();

    /* Update shared state */
    LWLockAcquire(AsyncQueueLock, LW_SHARED);
    QUEUE_BACKEND_POS(MyBackendId) = pos;
    advanceTail = QUEUE_POS_EQUAL(oldpos, QUEUE_TAIL);
    LWLockRelease(AsyncQueueLock);

    /* If we were the laziest backend, try to advance the tail pointer */
    if (advanceTail)
        asyncQueueAdvanceTail();

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

asyncQueueUnregister()

static void asyncQueueUnregister ( void  )

static

Definition at line 1222 of file async.c.

References amRegisteredListener, Assert, asyncQueueAdvanceTail(), InvalidOid, InvalidPid, LW_SHARED, LWLockAcquire(), LWLockRelease(), MyBackendId, MyProcPid, NIL, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_POS_EQUAL, and QUEUE_TAIL.

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

{
    bool        advanceTail;

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

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

    LWLockAcquire(AsyncQueueLock, LW_SHARED);
    /* check if entry is valid and oldest ... */
    advanceTail = (MyProcPid == QUEUE_BACKEND_PID(MyBackendId)) &&
        QUEUE_POS_EQUAL(QUEUE_BACKEND_POS(MyBackendId), QUEUE_TAIL);
    /* ... then mark it invalid */
    QUEUE_BACKEND_PID(MyBackendId) = InvalidPid;
    QUEUE_BACKEND_DBOID(MyBackendId) = InvalidOid;
    LWLockRelease(AsyncQueueLock);

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

    /* If we were the laziest backend, try to advance the tail pointer */
    if (advanceTail)
        asyncQueueAdvanceTail();
}

asyncQueueUsage()

static double asyncQueueUsage ( void  )

static

Definition at line 1466 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 472 of file async.c.

References add_size(), AsyncCtl, asyncQueuePagePrecedes(), i, InvalidOid, InvalidPid, AsyncQueueControl::lastQueueFillWarn, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LWTRANCHE_ASYNC_BUFFERS, MaxBackends, mul_size(), NUM_ASYNC_BUFFERS, offsetof, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_HEAD, QUEUE_POS_PAGE, QUEUE_TAIL, SET_QUEUE_POS, ShmemInitStruct(), SimpleLruInit(), SimpleLruWritePage(), SimpleLruZeroPage(), SlruScanDirCbDeleteAll(), and SlruScanDirectory().

Referenced by CreateSharedMemoryAndSemaphores().

{
    bool        found;
    int         slotno;
    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 */
        int         i;

        SET_QUEUE_POS(QUEUE_HEAD, 0, 0);
        SET_QUEUE_POS(QUEUE_TAIL, 0, 0);
        asyncQueueControl->lastQueueFillWarn = 0;
        /* zero'th entry won't be used, but let's initialize it anyway */
        for (i = 0; i <= MaxBackends; i++)
        {
            QUEUE_BACKEND_PID(i) = InvalidPid;
            QUEUE_BACKEND_DBOID(i) = InvalidOid;
            SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0);
        }
    }

    /*
     * Set up SLRU management of the pg_notify data.
     */
    AsyncCtl->PagePrecedes = asyncQueuePagePrecedes;
    SimpleLruInit(AsyncCtl, "async", NUM_ASYNC_BUFFERS, 0,
                  AsyncCtlLock, "pg_notify", LWTRANCHE_ASYNC_BUFFERS);
    /* Override default assumption that writes should be fsync'd */
    AsyncCtl->do_fsync = false;

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

        /* Now initialize page zero to empty */
        LWLockAcquire(AsyncCtlLock, LW_EXCLUSIVE);
        slotno = SimpleLruZeroPage(AsyncCtl, QUEUE_POS_PAGE(QUEUE_HEAD));
        /* This write is just to verify that pg_notify/ is writable */
        SimpleLruWritePage(AsyncCtl, slotno);
        LWLockRelease(AsyncCtlLock);
    }
}

AsyncShmemSize()

Size AsyncShmemSize

(

void

)

Definition at line 455 of file async.c.

References add_size(), MaxBackends, mul_size(), NUM_ASYNC_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_ASYNC_BUFFERS, 0));

    return size;
}

AtAbort_Notify()

void AtAbort_Notify

(

void

)

Definition at line 1622 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 911 of file async.c.

References ListenAction::action, 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 */
    foreach(p, pendingActions)
    {
        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 787 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 1729 of file async.c.

References GetCurrentTransactionNestLevel(), linitial, linitial_node, list_delete_first(), and list_length().

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.
     *
     * This routine could be called more than once at a given nesting level if
     * there is trouble during subxact abort.  Avoid dumping core by using
     * GetCurrentTransactionNestLevel as the indicator of how far we need to
     * prune the list.
     */
    while (list_length(upperPendingActions) > my_level - 2)
    {
        pendingActions = linitial_node(List, upperPendingActions);
        upperPendingActions = list_delete_first(upperPendingActions);
    }

    while (list_length(upperPendingNotifies) > my_level - 2)
    {
        pendingNotifies = (NotificationList *) linitial(upperPendingNotifies);
        upperPendingNotifies = list_delete_first(upperPendingNotifies);
    }
}

AtSubCommit_Notify()

void AtSubCommit_Notify

(

void

)

Definition at line 1672 of file async.c.

References AddEventToPendingNotifies(), Assert, AsyncExistsPendingNotify(), NotificationList::events, GetCurrentTransactionNestLevel(), lfirst, linitial, linitial_node, list_concat(), list_delete_first(), list_length(), and pendingNotifies.

Referenced by CommitSubTransaction().

{
    List       *parentPendingActions;
    NotificationList *parentPendingNotifies;

    parentPendingActions = linitial_node(List, upperPendingActions);
    upperPendingActions = list_delete_first(upperPendingActions);

    Assert(list_length(upperPendingActions) ==
           GetCurrentTransactionNestLevel() - 2);

    /*
     * Mustn't try to eliminate duplicates here --- see queue_listen()
     */
    pendingActions = list_concat(parentPendingActions, pendingActions);

    parentPendingNotifies = (NotificationList *) linitial(upperPendingNotifies);
    upperPendingNotifies = list_delete_first(upperPendingNotifies);

    Assert(list_length(upperPendingNotifies) ==
           GetCurrentTransactionNestLevel() - 2);

    if (pendingNotifies == NULL)
    {
        /* easy, no notify events happened in current subxact */
        pendingNotifies = parentPendingNotifies;
    }
    else if (parentPendingNotifies == NULL)
    {
        /* easy, subxact's list becomes parent's */
    }
    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 = parentPendingNotifies;
        /* 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);
        }
    }
}

AtSubStart_Notify()

void AtSubStart_Notify

(

void

)

Definition at line 1642 of file async.c.

References Assert, GetCurrentTransactionNestLevel(), lcons(), list_length(), MemoryContextSwitchTo(), NIL, and TopTransactionContext.

Referenced by StartSubTransaction().

{
    MemoryContext old_cxt;

    /* Keep the list-of-lists in TopTransactionContext for simplicity */
    old_cxt = MemoryContextSwitchTo(TopTransactionContext);

    upperPendingActions = lcons(pendingActions, upperPendingActions);

    Assert(list_length(upperPendingActions) ==
           GetCurrentTransactionNestLevel() - 1);

    pendingActions = NIL;

    upperPendingNotifies = lcons(pendingNotifies, upperPendingNotifies);

    Assert(list_length(upperPendingNotifies) ==
           GetCurrentTransactionNestLevel() - 1);

    pendingNotifies = NULL;

    MemoryContextSwitchTo(old_cxt);
}

ClearPendingActionsAndNotifies()

static void ClearPendingActionsAndNotifies ( void  )

static

Definition at line 2314 of file async.c.

References NIL.

Referenced by AtAbort_Notify(), and AtCommit_Notify().

{
    /*
     * We used to have to explicitly deallocate the list members and nodes,
     * because they were malloc'd.  Now, since we know they are palloc'd in
     * CurTransactionContext, we need not do that --- they'll go away
     * automatically at transaction exit.  We need only reset the list head
     * pointers.
     */
    pendingActions = NIL;
    pendingNotifies = NULL;
}

Exec_ListenCommit()

static void Exec_ListenCommit ( const char *  channel )

static

Definition at line 1046 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 958 of file async.c.

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

Referenced by PreCommit_Notify().

{
    QueuePosition head;
    QueuePosition max;
    int         i;

    /*
     * 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.)  But we only bother
     * with that if there's more than a page worth of notifications
     * outstanding, otherwise scanning all the other backends isn't worth it.
     *
     * We need exclusive lock here so we can look at other backends' entries.
     */
    LWLockAcquire(AsyncQueueLock, LW_EXCLUSIVE);
    head = QUEUE_HEAD;
    max = QUEUE_TAIL;
    if (QUEUE_POS_PAGE(max) != QUEUE_POS_PAGE(head))
    {
        for (i = 1; i <= MaxBackends; i++)
        {
            if (QUEUE_BACKEND_DBOID(i) == MyDatabaseId)
                max = QUEUE_POS_MAX(max, QUEUE_BACKEND_POS(i));
        }
    }
    QUEUE_BACKEND_POS(MyBackendId) = max;
    QUEUE_BACKEND_PID(MyBackendId) = MyProcPid;
    QUEUE_BACKEND_DBOID(MyBackendId) = MyDatabaseId;
    LWLockRelease(AsyncQueueLock);

    /* 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. Still, we could get notifications that
     * have already committed before we started to LISTEN.
     *
     * Note that we are not yet listening on anything, so we won't deliver any
     * notification to the frontend.  Also, although our transaction might
     * have executed NOTIFY, those message(s) aren't queued yet so we can't
     * see them in the queue.
     *
     * This will also advance the global tail pointer if possible.
     */
    if (!QUEUE_POS_EQUAL(max, head))
        asyncQueueReadAllNotifications();
}

Exec_UnlistenAllCommit()

static void Exec_UnlistenAllCommit ( void  )

static

Definition at line 1104 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 1073 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 1765 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 1203 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 2284 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 2298 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 2154 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 741 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 1449 of file async.c.

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

{
    double      usage;

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

    PG_RETURN_FLOAT8(usage);
}

pg_notify()

Datum pg_notify

(

PG_FUNCTION_ARGS

)

Definition at line 538 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 812 of file async.c.

References AccessExclusiveLock, ListenAction::action, 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 */
    foreach(p, pendingActions)
    {
        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 AsyncQueueLock.
         */
        (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 AsyncQueueLock 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(AsyncQueueLock, LW_EXCLUSIVE);
            asyncQueueFillWarning();
            if (asyncQueueIsFull())
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                         errmsg("too many notifications in the NOTIFY queue")));
            nextNotify = asyncQueueAddEntries(nextNotify);
            LWLockRelease(AsyncQueueLock);
        }
    }
}

ProcessCompletedNotifies()

void ProcessCompletedNotifies

(

void

)

Definition at line 1135 of file async.c.

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

Referenced by PostgresMain().

{
    MemoryContext caller_context;
    bool        signalled;

    /* 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 */
    signalled = SignalBackends();

    if (listenChannels != NIL)
    {
        /* Read the queue ourselves, and send relevant stuff to the frontend */
        asyncQueueReadAllNotifications();
    }
    else if (!signalled)
    {
        /*
         * If we found no other listening backends, and we aren't listening
         * ourselves, then we must execute asyncQueueAdvanceTail to flush the
         * queue, because ain't nobody else gonna do it.  This prevents queue
         * overflow when we're sending useless notifies to nobody. (A new
         * listener could have joined since we looked, but if so this is
         * harmless.)
         */
        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 2115 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", false);

    /*
     * 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", false);

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

ProcessNotifyInterrupt()

void ProcessNotifyInterrupt

(

void

)

Definition at line 1789 of file async.c.

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

Referenced by 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 659 of file async.c.

References generate_unaccent_rules::action, ListenAction::action, ListenAction::channel, CurTransactionContext, lappend(), MemoryContextSwitchTo(), offsetof, and palloc().

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

{
    MemoryContext oldcontext;
    ListenAction *actrec;

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

    pendingActions = lappend(pendingActions, actrec);

    MemoryContextSwitchTo(oldcontext);
}

SignalBackends()

static bool SignalBackends ( void  )

static

Definition at line 1550 of file async.c.

References DEBUG3, elog, i, InvalidPid, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MaxBackends, MyProcPid, palloc(), pfree(), PROCSIG_NOTIFY_INTERRUPT, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_HEAD, QUEUE_POS_EQUAL, and SendProcSignal().

Referenced by ProcessCompletedNotifies().

{
    bool        signalled = false;
    int32      *pids;
    BackendId  *ids;
    int         count;
    int         i;
    int32       pid;

    /*
     * Identify all backends that are listening and not already up-to-date. We
     * don't want to send signals while holding the AsyncQueueLock, so we just
     * build 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(AsyncQueueLock, LW_EXCLUSIVE);
    for (i = 1; i <= MaxBackends; i++)
    {
        pid = QUEUE_BACKEND_PID(i);
        if (pid != InvalidPid && pid != MyProcPid)
        {
            QueuePosition pos = QUEUE_BACKEND_POS(i);

            if (!QUEUE_POS_EQUAL(pos, QUEUE_HEAD))
            {
                pids[count] = pid;
                ids[count] = i;
                count++;
            }
        }
    }
    LWLockRelease(AsyncQueueLock);

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

        /*
         * Note: assuming things aren't broken, a signal failure here could
         * only occur if the target backend exited since we released
         * AsyncQueueLock; 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);
        else
            signalled = true;
    }

    pfree(pids);
    pfree(ids);

    return signalled;
}

Variable Documentation

amRegisteredListener

bool amRegisteredListener = false

static

Definition at line 390 of file async.c.

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

AsyncCtlData

SlruCtlData AsyncCtlData

static

Definition at line 266 of file async.c.

asyncQueueControl

AsyncQueueControl* asyncQueueControl

static

Definition at line 255 of file async.c.

backendHasSentNotifications

bool backendHasSentNotifications = false

static

Definition at line 393 of file async.c.

Referenced by PreCommit_Notify(), and ProcessCompletedNotifies().

listenChannels

List* listenChannels = NIL

static

Definition at line 296 of file async.c.

notifyInterruptPending

volatile sig_atomic_t notifyInterruptPending = false

Definition at line 384 of file async.c.

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

pendingActions

List* pendingActions = NIL

static

Definition at line 321 of file async.c.

pendingNotifies

NotificationList* pendingNotifies = NULL

static

Definition at line 373 of file async.c.

Referenced by AtSubCommit_Notify().

Trace_notify

bool Trace_notify = false

Definition at line 396 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 387 of file async.c.

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

upperPendingActions

List* upperPendingActions = NIL

static

Definition at line 323 of file async.c.

upperPendingNotifies

List* upperPendingNotifies = NIL

static

Definition at line 375 of file async.c.