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)

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

)

Definition at line 738 of file async.c.

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

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

Referenced by standard_ProcessUtility().

Async_Notify()

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

Definition at line 591 of file async.c.

{
    int         my_level = GetCurrentTransactionNestLevel();
    size_t      channel_len;
    size_t      payload_len;
    Notification *n;
    MemoryContext oldcontext;
 
    if (IsParallelWorker())
        elog(ERROR, "cannot send notifications from a parallel worker");
 
    if (Trace_notify)
        elog(DEBUG1, "Async_Notify(%s)", channel);
 
    channel_len = channel ? strlen(channel) : 0;
    payload_len = payload ? strlen(payload) : 0;
 
    /* a channel name must be specified */
    if (channel_len == 0)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("channel name cannot be empty")));
 
    /* enforce length limits */
    if (channel_len >= NAMEDATALEN)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("channel name too long")));
 
    if (payload_len >= NOTIFY_PAYLOAD_MAX_LENGTH)
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("payload string too long")));
 
    /*
     * We must construct the Notification entry, even if we end up not using
     * it, in order to compare it cheaply to existing list entries.
     *
     * The notification list needs to live until end of transaction, so store
     * it in the transaction context.
     */
    oldcontext = MemoryContextSwitchTo(CurTransactionContext);
 
    n = (Notification *) palloc(offsetof(Notification, data) +
                                channel_len + payload_len + 2);
    n->channel_len = channel_len;
    n->payload_len = payload_len;
    strcpy(n->data, channel);
    if (payload)
        strcpy(n->data + channel_len + 1, payload);
    else
        n->data[channel_len + 1] = '\0';
 
    if (pendingNotifies == NULL || my_level > pendingNotifies->nestingLevel)
    {
        NotificationList *notifies;
 
        /*
         * First notify event in current (sub)xact. Note that we allocate the
         * NotificationList in TopTransactionContext; the nestingLevel might
         * get changed later by AtSubCommit_Notify.
         */
        notifies = (NotificationList *)
            MemoryContextAlloc(TopTransactionContext,
                               sizeof(NotificationList));
        notifies->nestingLevel = my_level;
        notifies->events = list_make1(n);
        /* We certainly don't need a hashtable yet */
        notifies->hashtab = NULL;
        notifies->upper = pendingNotifies;
        pendingNotifies = notifies;
    }
    else
    {
        /* Now check for duplicates */
        if (AsyncExistsPendingNotify(n))
        {
            /* It's a dup, so forget it */
            pfree(n);
            MemoryContextSwitchTo(oldcontext);
            return;
        }
 
        /* Append more events to existing list */
        AddEventToPendingNotifies(n);
    }
 
    MemoryContextSwitchTo(oldcontext);
}

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

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

Async_Unlisten()

void Async_Unlisten

(

const char *

channel

)

Definition at line 752 of file async.c.

{
    if (Trace_notify)
        elog(DEBUG1, "Async_Unlisten(%s,%d)", channel, MyProcPid);
 
    /* If we couldn't possibly be listening, no need to queue anything */
    if (pendingActions == NULL && !unlistenExitRegistered)
        return;
 
    queue_listen(LISTEN_UNLISTEN, channel);
}

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

Referenced by standard_ProcessUtility().

Async_UnlistenAll()

void Async_UnlistenAll

(

void

)

Definition at line 770 of file async.c.

{
    if (Trace_notify)
        elog(DEBUG1, "Async_UnlistenAll(%d)", MyProcPid);
 
    /* If we couldn't possibly be listening, no need to queue anything */
    if (pendingActions == NULL && !unlistenExitRegistered)
        return;
 
    queue_listen(LISTEN_UNLISTEN_ALL, "");
}

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

Referenced by DiscardAll(), and standard_ProcessUtility().

AsyncShmemInit()

void AsyncShmemInit

(

void

)

Definition at line 502 of file async.c.

{
    bool        found;
    Size        size;
 
    /*
     * Create or attach to the AsyncQueueControl structure.
     */
    size = mul_size(MaxBackends, sizeof(QueueBackendStatus));
    size = add_size(size, offsetof(AsyncQueueControl, backend));
 
    asyncQueueControl = (AsyncQueueControl *)
        ShmemInitStruct("Async Queue Control", size, &found);
 
    if (!found)
    {
        /* First time through, so initialize it */
        SET_QUEUE_POS(QUEUE_HEAD, 0, 0);
        SET_QUEUE_POS(QUEUE_TAIL, 0, 0);
        QUEUE_STOP_PAGE = 0;
        QUEUE_FIRST_LISTENER = INVALID_PROC_NUMBER;
        asyncQueueControl->lastQueueFillWarn = 0;
        for (int i = 0; i < MaxBackends; i++)
        {
            QUEUE_BACKEND_PID(i) = InvalidPid;
            QUEUE_BACKEND_DBOID(i) = InvalidOid;
            QUEUE_NEXT_LISTENER(i) = INVALID_PROC_NUMBER;
            SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0);
        }
    }
 
    /*
     * Set up SLRU management of the pg_notify data. Note that long segment
     * names are used in order to avoid wraparound.
     */
    NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
    SimpleLruInit(NotifyCtl, "notify", notify_buffers, 0,
                  "pg_notify", LWTRANCHE_NOTIFY_BUFFER, LWTRANCHE_NOTIFY_SLRU,
                  SYNC_HANDLER_NONE, true);
 
    if (!found)
    {
        /*
         * During start or reboot, clean out the pg_notify directory.
         */
        (void) SlruScanDirectory(NotifyCtl, SlruScanDirCbDeleteAll, NULL);
    }
}

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

Referenced by CreateOrAttachShmemStructs().

AsyncShmemSize()

Size AsyncShmemSize

(

void

)

Definition at line 485 of file async.c.

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

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

Referenced by CalculateShmemSize().

AtAbort_Notify()

void AtAbort_Notify

(

void

)

Definition at line 1671 of file async.c.

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

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

Referenced by AbortTransaction().

AtCommit_Notify()

void AtCommit_Notify

(

void

)

Definition at line 968 of file async.c.

{
    ListCell   *p;
 
    /*
     * Allow transactions that have not executed LISTEN/UNLISTEN/NOTIFY to
     * return as soon as possible
     */
    if (!pendingActions && !pendingNotifies)
        return;
 
    if (Trace_notify)
        elog(DEBUG1, "AtCommit_Notify");
 
    /* Perform any pending listen/unlisten actions */
    if (pendingActions != NULL)
    {
        foreach(p, pendingActions->actions)
        {
            ListenAction *actrec = (ListenAction *) lfirst(p);
 
            switch (actrec->action)
            {
                case LISTEN_LISTEN:
                    Exec_ListenCommit(actrec->channel);
                    break;
                case LISTEN_UNLISTEN:
                    Exec_UnlistenCommit(actrec->channel);
                    break;
                case LISTEN_UNLISTEN_ALL:
                    Exec_UnlistenAllCommit();
                    break;
            }
        }
    }
 
    /* If no longer listening to anything, get out of listener array */
    if (amRegisteredListener && listenChannels == NIL)
        asyncQueueUnregister();
 
    /*
     * Send signals to listening backends.  We need do this only if there are
     * pending notifies, which were previously added to the shared queue by
     * PreCommit_Notify().
     */
    if (pendingNotifies != NULL)
        SignalBackends();
 
    /*
     * If it's time to try to advance the global tail pointer, do that.
     *
     * (It might seem odd to do this in the sender, when more than likely the
     * listeners won't yet have read the messages we just sent.  However,
     * there's less contention if only the sender does it, and there is little
     * need for urgency in advancing the global tail.  So this typically will
     * be clearing out messages that were sent some time ago.)
     */
    if (tryAdvanceTail)
    {
        tryAdvanceTail = false;
        asyncQueueAdvanceTail();
    }
 
    /* And clean up */
    ClearPendingActionsAndNotifies();
}

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

Referenced by CommitTransaction().

AtPrepare_Notify()

void AtPrepare_Notify

(

void

)

Definition at line 836 of file async.c.

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

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

Referenced by PrepareTransaction().

AtSubAbort_Notify()

void AtSubAbort_Notify

(

void

)

Definition at line 1761 of file async.c.

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

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

Referenced by AbortSubTransaction().

AtSubCommit_Notify()

void AtSubCommit_Notify

(

void

)

Definition at line 1691 of file async.c.

{
    int         my_level = GetCurrentTransactionNestLevel();
 
    /* If there are actions at our nesting level, we must reparent them. */
    if (pendingActions != NULL &&
        pendingActions->nestingLevel >= my_level)
    {
        if (pendingActions->upper == NULL ||
            pendingActions->upper->nestingLevel < my_level - 1)
        {
            /* nothing to merge; give the whole thing to the parent */
            --pendingActions->nestingLevel;
        }
        else
        {
            ActionList *childPendingActions = pendingActions;
 
            pendingActions = pendingActions->upper;
 
            /*
             * Mustn't try to eliminate duplicates here --- see queue_listen()
             */
            pendingActions->actions =
                list_concat(pendingActions->actions,
                            childPendingActions->actions);
            pfree(childPendingActions);
        }
    }
 
    /* If there are notifies at our nesting level, we must reparent them. */
    if (pendingNotifies != NULL &&
        pendingNotifies->nestingLevel >= my_level)
    {
        Assert(pendingNotifies->nestingLevel == my_level);
 
        if (pendingNotifies->upper == NULL ||
            pendingNotifies->upper->nestingLevel < my_level - 1)
        {
            /* nothing to merge; give the whole thing to the parent */
            --pendingNotifies->nestingLevel;
        }
        else
        {
            /*
             * Formerly, we didn't bother to eliminate duplicates here, but
             * now we must, else we fall foul of "Assert(!found)", either here
             * or during a later attempt to build the parent-level hashtable.
             */
            NotificationList *childPendingNotifies = pendingNotifies;
            ListCell   *l;
 
            pendingNotifies = pendingNotifies->upper;
            /* Insert all the subxact's events into parent, except for dups */
            foreach(l, childPendingNotifies->events)
            {
                Notification *childn = (Notification *) lfirst(l);
 
                if (!AsyncExistsPendingNotify(childn))
                    AddEventToPendingNotifies(childn);
            }
            pfree(childPendingNotifies);
        }
    }
}

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

Referenced by CommitSubTransaction().

HandleNotifyInterrupt()

void HandleNotifyInterrupt

(

void

)

Definition at line 1804 of file async.c.

{
    /*
     * Note: this is called by a SIGNAL HANDLER. You must be very wary what
     * you do here.
     */
 
    /* signal that work needs to be done */
    notifyInterruptPending = true;
 
    /* make sure the event is processed in due course */
    SetLatch(MyLatch);
}

References MyLatch, notifyInterruptPending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

NotifyMyFrontEnd()

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

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

PreCommit_Notify()

void PreCommit_Notify

(

void

)

Definition at line 861 of file async.c.

{
    ListCell   *p;
 
    if (!pendingActions && !pendingNotifies)
        return;                 /* no relevant statements in this xact */
 
    if (Trace_notify)
        elog(DEBUG1, "PreCommit_Notify");
 
    /* Preflight for any pending listen/unlisten actions */
    if (pendingActions != NULL)
    {
        foreach(p, pendingActions->actions)
        {
            ListenAction *actrec = (ListenAction *) lfirst(p);
 
            switch (actrec->action)
            {
                case LISTEN_LISTEN:
                    Exec_ListenPreCommit();
                    break;
                case LISTEN_UNLISTEN:
                    /* there is no Exec_UnlistenPreCommit() */
                    break;
                case LISTEN_UNLISTEN_ALL:
                    /* there is no Exec_UnlistenAllPreCommit() */
                    break;
            }
        }
    }
 
    /* Queue any pending notifies (must happen after the above) */
    if (pendingNotifies)
    {
        ListCell   *nextNotify;
 
        /*
         * Make sure that we have an XID assigned to the current transaction.
         * GetCurrentTransactionId is cheap if we already have an XID, but not
         * so cheap if we don't, and we'd prefer not to do that work while
         * holding NotifyQueueLock.
         */
        (void) GetCurrentTransactionId();
 
        /*
         * Serialize writers by acquiring a special lock that we hold till
         * after commit.  This ensures that queue entries appear in commit
         * order, and in particular that there are never uncommitted queue
         * entries ahead of committed ones, so an uncommitted transaction
         * can't block delivery of deliverable notifications.
         *
         * We use a heavyweight lock so that it'll automatically be released
         * after either commit or abort.  This also allows deadlocks to be
         * detected, though really a deadlock shouldn't be possible here.
         *
         * The lock is on "database 0", which is pretty ugly but it doesn't
         * seem worth inventing a special locktag category just for this.
         * (Historical note: before PG 9.0, a similar lock on "database 0" was
         * used by the flatfiles mechanism.)
         */
        LockSharedObject(DatabaseRelationId, InvalidOid, 0,
                         AccessExclusiveLock);
 
        /* Now push the notifications into the queue */
        nextNotify = list_head(pendingNotifies->events);
        while (nextNotify != NULL)
        {
            /*
             * Add the pending notifications to the queue.  We acquire and
             * release NotifyQueueLock once per page, which might be overkill
             * but it does allow readers to get in while we're doing this.
             *
             * A full queue is very uncommon and should really not happen,
             * given that we have so much space available in the SLRU pages.
             * Nevertheless we need to deal with this possibility. Note that
             * when we get here we are in the process of committing our
             * transaction, but we have not yet committed to clog, so at this
             * point in time we can still roll the transaction back.
             */
            LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
            asyncQueueFillWarning();
            if (asyncQueueIsFull())
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                         errmsg("too many notifications in the NOTIFY queue")));
            nextNotify = asyncQueueAddEntries(nextNotify);
            LWLockRelease(NotifyQueueLock);
        }
 
        /* Note that we don't clear pendingNotifies; AtCommit_Notify will. */
    }
}

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

Referenced by CommitTransaction().

ProcessNotifyInterrupt()

void ProcessNotifyInterrupt

(

bool

flush

)

Definition at line 1834 of file async.c.

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

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

Referenced by PostgresMain(), and ProcessClientReadInterrupt().

Variable Documentation

max_notify_queue_pages

PGDLLIMPORT int max_notify_queue_pages

extern

Definition at line 428 of file async.c.

Referenced by asyncQueueIsFull(), and asyncQueueUsage().

notifyInterruptPending

PGDLLIMPORT volatile sig_atomic_t notifyInterruptPending

extern

Definition at line 413 of file async.c.

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

Trace_notify

PGDLLIMPORT bool Trace_notify

extern

Definition at line 425 of file async.c.

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