#include <signal.h>

Include dependency graph for async.h:

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

Macros
#define	NUM_NOTIFY_BUFFERS 8

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 volatile sig_atomic_t	notifyInterruptPending

Macro Definition Documentation

◆ NUM_NOTIFY_BUFFERS

#define NUM_NOTIFY_BUFFERS 8

Definition at line 21 of file async.h.

Function Documentation

◆ Async_Listen()

void Async_Listen ( const char * channel )

Definition at line 772 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 625 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 786 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 804 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 533 of file async.c.

 {
     bool        found;
     Size        size;
  
     /*
      * Create or attach to the AsyncQueueControl structure.
      *
      * The used entries in the backend[] array run from 1 to MaxBackends; the
      * zero'th entry is unused but must be allocated.
      */
     size = mul_size(MaxBackends + 1, sizeof(QueueBackendStatus));
     size = add_size(size, offsetof(AsyncQueueControl, backend));
  
     asyncQueueControl = (AsyncQueueControl *)
         ShmemInitStruct("Async Queue Control", size, &found);
  
     if (!found)
     {
         /* First time through, so initialize it */
         SET_QUEUE_POS(QUEUE_HEAD, 0, 0);
         SET_QUEUE_POS(QUEUE_TAIL, 0, 0);
         QUEUE_STOP_PAGE = 0;
         QUEUE_FIRST_LISTENER = InvalidBackendId;
         asyncQueueControl->lastQueueFillWarn = 0;
         /* zero'th entry won't be used, but let's initialize it anyway */
         for (int i = 0; i <= MaxBackends; i++)
         {
             QUEUE_BACKEND_PID(i) = InvalidPid;
             QUEUE_BACKEND_DBOID(i) = InvalidOid;
             QUEUE_NEXT_LISTENER(i) = InvalidBackendId;
             SET_QUEUE_POS(QUEUE_BACKEND_POS(i), 0, 0);
         }
     }
  
     /*
      * Set up SLRU management of the pg_notify data.
      */
     NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
     SimpleLruInit(NotifyCtl, "Notify", NUM_NOTIFY_BUFFERS, 0,
                   NotifySLRULock, "pg_notify", LWTRANCHE_NOTIFY_BUFFER,
                   SYNC_HANDLER_NONE);
  
     if (!found)
     {
         /*
          * During start or reboot, clean out the pg_notify directory.
          */
         (void) SlruScanDirectory(NotifyCtl, SlruScanDirCbDeleteAll, NULL);
     }
 }

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

Referenced by CreateSharedMemoryAndSemaphores().

◆ AsyncShmemSize()

Size AsyncShmemSize ( void )

Definition at line 516 of file async.c.

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

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

Referenced by CalculateShmemSize().

◆ AtAbort_Notify()

void AtAbort_Notify ( void )

Definition at line 1720 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 1002 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 870 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 1810 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 1740 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 1853 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 2278 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 895 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 1883 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

◆ notifyInterruptPending

PGDLLIMPORT volatile sig_atomic_t notifyInterruptPending

extern

Definition at line 431 of file async.c.

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

◆ Trace_notify

PGDLLIMPORT bool Trace_notify