async.h File Reference

#include <signal.h>

Include dependency graph for async.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
Size	AsyncShmemSize (void)
void	AsyncShmemInit (void)
void	NotifyMyFrontEnd (const char *channel, const char *payload, int32 srcPid)
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)
void	PreCommit_Notify (void)
void	AtCommit_Notify (void)
void	AtAbort_Notify (void)
void	AtSubCommit_Notify (void)
void	AtSubAbort_Notify (void)
void	AtPrepare_Notify (void)
void	HandleNotifyInterrupt (void)
void	ProcessNotifyInterrupt (bool flush)
void	AsyncNotifyFreezeXids (TransactionId newFrozenXid)

Variables
PGDLLIMPORT bool	Trace_notify
PGDLLIMPORT int	max_notify_queue_pages
PGDLLIMPORT volatile sig_atomic_t	notifyInterruptPending

Function Documentation

Async_Listen()

void Async_Listen ( const char *  channel )

extern

Definition at line 1046 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  )

extern

Definition at line 897 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;
        /* We won't build uniqueChannelNames/Hash till later, either */
        notifies->uniqueChannelNames = NIL;
        notifies->uniqueChannelHash = 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, fb(), GetCurrentTransactionNestLevel(), IsParallelWorker, list_make1, MemoryContextAlloc(), MemoryContextSwitchTo(), NAMEDATALEN, NotificationList::nestingLevel, NIL, NOTIFY_PAYLOAD_MAX_LENGTH, palloc(), Notification::payload_len, pendingNotifies, pfree(), TopTransactionContext, and Trace_notify.

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

Async_Unlisten()

void Async_Unlisten ( const char *  channel )

extern

Definition at line 1060 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, fb(), LISTEN_UNLISTEN, MyProcPid, pendingActions, queue_listen(), Trace_notify, and unlistenExitRegistered.

Referenced by standard_ProcessUtility().

Async_UnlistenAll()

void Async_UnlistenAll ( void  )

extern

Definition at line 1078 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, fb(), LISTEN_UNLISTEN_ALL, MyProcPid, pendingActions, queue_listen(), Trace_notify, and unlistenExitRegistered.

Referenced by DiscardAll(), and standard_ProcessUtility().

AsyncNotifyFreezeXids()

void AsyncNotifyFreezeXids ( TransactionId  newFrozenXid )

extern

Definition at line 2954 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, &pos);
            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(), fb(), FrozenTransactionId, InvalidTransactionId, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), NotifyCtl, QUEUE_HEAD, QUEUE_POS_EQUAL, QUEUE_POS_OFFSET, QUEUE_POS_PAGE, QUEUE_TAIL, SimpleLruGetBankLock(), SimpleLruReadPage(), TransactionIdDidCommit(), TransactionIdIsNormal, and TransactionIdPrecedes().

Referenced by vac_truncate_clog().

AsyncShmemInit()

void AsyncShmemInit ( void  )

extern

Definition at line 803 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;
        asyncQueueControl->globalChannelTableDSA = DSA_HANDLE_INVALID;
        asyncQueueControl->globalChannelTableDSH = DSHASH_HANDLE_INVALID;
        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);
            QUEUE_BACKEND_WAKEUP_PENDING(i) = false;
            QUEUE_BACKEND_IS_ADVANCING(i) = false;
        }
    }
 
    /*
     * Set up SLRU management of the pg_notify data. Note that long segment
     * names are used in order to avoid wraparound.
     */
    NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
    NotifyCtl->errdetail_for_io_error = asyncQueueErrdetailForIoError;
    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, asyncQueueErrdetailForIoError(), asyncQueuePagePrecedes(), DSA_HANDLE_INVALID, DSHASH_HANDLE_INVALID, fb(), AsyncQueueControl::globalChannelTableDSA, AsyncQueueControl::globalChannelTableDSH, i, INVALID_PROC_NUMBER, InvalidOid, InvalidPid, AsyncQueueControl::lastQueueFillWarn, MaxBackends, mul_size(), notify_buffers, NotifyCtl, QUEUE_BACKEND_DBOID, QUEUE_BACKEND_IS_ADVANCING, QUEUE_BACKEND_PID, QUEUE_BACKEND_POS, QUEUE_BACKEND_WAKEUP_PENDING, 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  )

extern

Definition at line 786 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(), fb(), MaxBackends, mul_size(), notify_buffers, and SimpleLruShmemSize().

Referenced by CalculateShmemSize().

AtAbort_Notify()

void AtAbort_Notify ( void  )

extern

Definition at line 2421 of file async.c.

{
    /* Revert staged listen/unlisten changes */
    ApplyPendingListenActions(false);
 
    /* If we're no longer listening on anything, unregister */
    if (amRegisteredListener && LocalChannelTableIsEmpty())
        asyncQueueUnregister();
 
    /* And clean up */
    ClearPendingActionsAndNotifies();
}

References amRegisteredListener, ApplyPendingListenActions(), asyncQueueUnregister(), ClearPendingActionsAndNotifies(), and LocalChannelTableIsEmpty.

Referenced by AbortTransaction().

AtCommit_Notify()

void AtCommit_Notify ( void  )

extern

Definition at line 1381 of file async.c.

{
    /*
     * 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");
 
    /* Apply staged listen/unlisten changes */
    ApplyPendingListenActions(true);
 
    /* If no longer listening to anything, get out of listener array */
    if (amRegisteredListener && LocalChannelTableIsEmpty())
        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 amRegisteredListener, ApplyPendingListenActions(), asyncQueueAdvanceTail(), asyncQueueUnregister(), ClearPendingActionsAndNotifies(), DEBUG1, elog, fb(), LocalChannelTableIsEmpty, pendingActions, pendingNotifies, SignalBackends(), Trace_notify, and tryAdvanceTail.

Referenced by CommitTransaction().

AtPrepare_Notify()

void AtPrepare_Notify ( void  )

extern

Definition at line 1163 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, fb(), pendingActions, and pendingNotifies.

Referenced by PrepareTransaction().

AtSubAbort_Notify()

void AtSubAbort_Notify ( void  )

extern

Definition at line 2510 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 fb(), GetCurrentTransactionNestLevel(), ActionList::nestingLevel, NotificationList::nestingLevel, pendingActions, pendingNotifies, pfree(), ActionList::upper, and NotificationList::upper.

Referenced by AbortSubTransaction().

AtSubCommit_Notify()

void AtSubCommit_Notify ( void  )

extern

Definition at line 2440 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(), fb(), GetCurrentTransactionNestLevel(), lfirst, list_concat(), ActionList::nestingLevel, NotificationList::nestingLevel, pendingActions, pendingNotifies, pfree(), ActionList::upper, and NotificationList::upper.

Referenced by CommitSubTransaction().

HandleNotifyInterrupt()

void HandleNotifyInterrupt ( void  )

extern

Definition at line 2553 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().

NotifyMyFrontEnd()

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

extern

Definition at line 3101 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().

PreCommit_Notify()

void PreCommit_Notify ( void  )

extern

Definition at line 1188 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 */
    initGlobalChannelTable();
 
    if (pendingActions != NULL)
    {
        /* Ensure we have a local channel table */
        initLocalChannelTable();
        /* Create pendingListenActions hash table for this transaction */
        initPendingListenActions();
 
        /* Stage all the actions this transaction wants to perform */
        foreach(p, pendingActions->actions)
        {
            ListenAction *actrec = (ListenAction *) lfirst(p);
 
            switch (actrec->action)
            {
                case LISTEN_LISTEN:
                    BecomeRegisteredListener();
                    PrepareTableEntriesForListen(actrec->channel);
                    break;
                case LISTEN_UNLISTEN:
                    PrepareTableEntriesForUnlisten(actrec->channel);
                    break;
                case LISTEN_UNLISTEN_ALL:
                    PrepareTableEntriesForUnlistenAll();
                    break;
            }
        }
    }
 
    /* Queue any pending notifies (must happen after the above) */
    if (pendingNotifies)
    {
        ListCell   *nextNotify;
        bool        firstIteration = true;
 
        /*
         * Build list of unique channel names being notified for use by
         * SignalBackends().
         *
         * If uniqueChannelHash is available, use it to efficiently get the
         * unique channels.  Otherwise, fall back to the O(N^2) approach.
         */
        pendingNotifies->uniqueChannelNames = NIL;
        if (pendingNotifies->uniqueChannelHash != NULL)
        {
            HASH_SEQ_STATUS status;
            ChannelName *channelEntry;
 
            hash_seq_init(&status, pendingNotifies->uniqueChannelHash);
            while ((channelEntry = (ChannelName *) hash_seq_search(&status)) != NULL)
                pendingNotifies->uniqueChannelNames =
                    lappend(pendingNotifies->uniqueChannelNames,
                            channelEntry->channel);
        }
        else
        {
            /* O(N^2) approach is better for small number of notifications */
            foreach_ptr(Notification, n, pendingNotifies->events)
            {
                char       *channel = n->data;
                bool        found = false;
 
                /* Name present in list? */
                foreach_ptr(char, oldchan, pendingNotifies->uniqueChannelNames)
                {
                    if (strcmp(oldchan, channel) == 0)
                    {
                        found = true;
                        break;
                    }
                }
                /* Add if not already in list */
                if (!found)
                    pendingNotifies->uniqueChannelNames =
                        lappend(pendingNotifies->uniqueChannelNames,
                                channel);
            }
        }
 
        /* Preallocate workspace that will be needed by SignalBackends() */
        if (signalPids == NULL)
            signalPids = MemoryContextAlloc(TopMemoryContext,
                                            MaxBackends * sizeof(int32));
 
        if (signalProcnos == NULL)
            signalProcnos = MemoryContextAlloc(TopMemoryContext,
                                               MaxBackends * sizeof(ProcNumber));
 
        /*
         * 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);
 
        /*
         * For the direct advancement optimization in SignalBackends(), we
         * need to ensure that no other backend can insert queue entries
         * between queueHeadBeforeWrite and queueHeadAfterWrite.  The
         * heavyweight lock above provides this guarantee, since it serializes
         * all writers.
         *
         * Note: if the heavyweight lock were ever removed for scalability
         * reasons, we could achieve the same guarantee by holding
         * NotifyQueueLock in EXCLUSIVE mode across all our insertions, rather
         * than releasing and reacquiring it for each page as we do below.
         */
 
        /* Initialize values to a safe default in case list is empty */
        SET_QUEUE_POS(queueHeadBeforeWrite, 0, 0);
        SET_QUEUE_POS(queueHeadAfterWrite, 0, 0);
 
        /* 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);
            if (firstIteration)
            {
                queueHeadBeforeWrite = QUEUE_HEAD;
                firstIteration = false;
            }
            asyncQueueFillWarning();
            if (asyncQueueIsFull())
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                         errmsg("too many notifications in the NOTIFY queue")));
            nextNotify = asyncQueueAddEntries(nextNotify);
            queueHeadAfterWrite = QUEUE_HEAD;
            LWLockRelease(NotifyQueueLock);
        }
 
        /* Note that we don't clear pendingNotifies; AtCommit_Notify will. */
    }
}

References AccessExclusiveLock, ActionList::actions, asyncQueueAddEntries(), asyncQueueFillWarning(), asyncQueueIsFull(), BecomeRegisteredListener(), DEBUG1, elog, ereport, errcode(), errmsg, ERROR, NotificationList::events, fb(), foreach_ptr, GetCurrentTransactionId(), hash_seq_init(), hash_seq_search(), initGlobalChannelTable(), initLocalChannelTable(), initPendingListenActions(), InvalidOid, lappend(), lfirst, list_head(), LISTEN_LISTEN, LISTEN_UNLISTEN, LISTEN_UNLISTEN_ALL, LockSharedObject(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MaxBackends, MemoryContextAlloc(), NIL, pendingActions, pendingNotifies, PrepareTableEntriesForListen(), PrepareTableEntriesForUnlisten(), PrepareTableEntriesForUnlistenAll(), QUEUE_HEAD, queueHeadAfterWrite, queueHeadBeforeWrite, SET_QUEUE_POS, signalPids, signalProcnos, TopMemoryContext, Trace_notify, NotificationList::uniqueChannelHash, and NotificationList::uniqueChannelNames.

Referenced by CommitTransaction().

ProcessNotifyInterrupt()

void ProcessNotifyInterrupt ( bool  flush )

extern

Definition at line 2583 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().

Variable Documentation

max_notify_queue_pages

PGDLLIMPORT int max_notify_queue_pages

extern

Definition at line 570 of file async.c.

Referenced by asyncQueueIsFull(), and asyncQueueUsage().

notifyInterruptPending

PGDLLIMPORT volatile sig_atomic_t notifyInterruptPending

extern

Definition at line 538 of file async.c.

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

Trace_notify

PGDLLIMPORT bool Trace_notify

extern

Definition at line 567 of file async.c.

Referenced by Async_Listen(), Async_Notify(), Async_Unlisten(), Async_UnlistenAll(), AtCommit_Notify(), BecomeRegisteredListener(), CleanupListenersOnExit(), PreCommit_Notify(), and ProcessIncomingNotify().