From b99e3461eccc1590ab4720e06116c948ded8ea99 Mon Sep 17 00:00:00 2001 From: Robert Haas Date: Tue, 10 Sep 2019 12:31:52 -0400 Subject: [PATCH] Remove AtSubStart_Notify. Allocate notify-related state lazily instead. --- src/backend/access/transam/xact.c | 1 - src/backend/commands/async.c | 297 +++++++++++++++++------------- src/include/commands/async.h | 1 - 3 files changed, 164 insertions(+), 135 deletions(-) diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 9162286c98..fc55fa6d53 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -4743,7 +4743,6 @@ StartSubTransaction(void) */ AtSubStart_Memory(); AtSubStart_ResourceOwner(); - AtSubStart_Notify(); AfterTriggerBeginSubXact(); s->state = TRANS_INPROGRESS; diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index 6cb2d445f0..36908c384d 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -318,9 +318,14 @@ typedef struct char channel[FLEXIBLE_ARRAY_MEMBER]; /* nul-terminated string */ } ListenAction; -static List *pendingActions = NIL; /* list of ListenAction */ +typedef struct ActionList +{ + int nestingLevel; /* current transaction nesting depth */ + List *actions; /* list of ListenAction structs */ + struct ActionList *upper; /* details for upper transaction levels */ +} ActionList; -static List *upperPendingActions = NIL; /* list of upper-xact lists */ +static ActionList *pendingActions = NULL; /* * State for outbound notifies consists of a list of all channels+payloads @@ -359,8 +364,10 @@ typedef struct Notification typedef struct NotificationList { + int nestingLevel; /* current transaction nesting depth */ List *events; /* list of Notification structs */ HTAB *hashtab; /* hash of NotificationHash structs, or NULL */ + struct NotificationList *upper; /* details for upper transaction levels */ } NotificationList; #define MIN_HASHABLE_NOTIFIES 16 /* threshold to build hashtab */ @@ -370,9 +377,7 @@ typedef struct NotificationHash Notification *event; /* => the actual Notification struct */ } NotificationHash; -static NotificationList *pendingNotifies = NULL; /* current list, if any */ - -static List *upperPendingNotifies = NIL; /* list of upper-xact lists */ +static NotificationList *pendingNotifies = NULL; /* * Inbound notifications are initially processed by HandleNotifyInterrupt(), @@ -571,6 +576,7 @@ pg_notify(PG_FUNCTION_ARGS) void Async_Notify(const char *channel, const char *payload) { + int my_level = GetCurrentTransactionNestLevel(); size_t channel_len; size_t payload_len; Notification *n; @@ -621,25 +627,36 @@ Async_Notify(const char *channel, const char *payload) else n->data[channel_len + 1] = '\0'; - /* Now check for duplicates */ - if (AsyncExistsPendingNotify(n)) + if (pendingNotifies == NULL || my_level > pendingNotifies->nestingLevel) { - /* It's a dup, so forget it */ - pfree(n); - MemoryContextSwitchTo(oldcontext); - return; - } + NotificationList *notifies; - if (pendingNotifies == NULL) - { - /* First notify event in current (sub)xact */ - pendingNotifies = (NotificationList *) palloc(sizeof(NotificationList)); - pendingNotifies->events = list_make1(n); + /* + * 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 */ - pendingNotifies->hashtab = NULL; + 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); } @@ -660,6 +677,7 @@ queue_listen(ListenActionKind action, const char *channel) { MemoryContext oldcontext; ListenAction *actrec; + int my_level = GetCurrentTransactionNestLevel(); /* * Unlike Async_Notify, we don't try to collapse out duplicates. It would @@ -675,7 +693,24 @@ queue_listen(ListenActionKind action, const char *channel) actrec->action = action; strcpy(actrec->channel, channel); - pendingActions = lappend(pendingActions, actrec); + if (pendingActions == NULL || my_level > pendingActions->nestingLevel) + { + ActionList *actions; + + /* + * First action in current sub(xact). Note that we allocate the + * ActionList in TopTransactionContext; the nestingLevel might get + * changed later by AtSubCommit_Notify. + */ + actions = (ActionList *) + MemoryContextAlloc(TopTransactionContext, sizeof(ActionList)); + actions->nestingLevel = my_level; + actions->actions = list_make1(actrec); + actions->upper = pendingActions; + pendingActions = actions; + } + else + pendingActions->actions = lappend(pendingActions->actions, actrec); MemoryContextSwitchTo(oldcontext); } @@ -706,7 +741,7 @@ Async_Unlisten(const char *channel) elog(DEBUG1, "Async_Unlisten(%s,%d)", channel, MyProcPid); /* If we couldn't possibly be listening, no need to queue anything */ - if (pendingActions == NIL && !unlistenExitRegistered) + if (pendingActions == NULL && !unlistenExitRegistered) return; queue_listen(LISTEN_UNLISTEN, channel); @@ -724,7 +759,7 @@ Async_UnlistenAll(void) elog(DEBUG1, "Async_UnlistenAll(%d)", MyProcPid); /* If we couldn't possibly be listening, no need to queue anything */ - if (pendingActions == NIL && !unlistenExitRegistered) + if (pendingActions == NULL && !unlistenExitRegistered) return; queue_listen(LISTEN_UNLISTEN_ALL, ""); @@ -820,21 +855,24 @@ PreCommit_Notify(void) elog(DEBUG1, "PreCommit_Notify"); /* Preflight for any pending listen/unlisten actions */ - foreach(p, pendingActions) + if (pendingActions != NULL) { - ListenAction *actrec = (ListenAction *) lfirst(p); - - switch (actrec->action) + foreach(p, pendingActions->actions) { - 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; + 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; + } } } @@ -923,21 +961,24 @@ AtCommit_Notify(void) elog(DEBUG1, "AtCommit_Notify"); /* Perform any pending listen/unlisten actions */ - foreach(p, pendingActions) + if (pendingActions != NULL) { - ListenAction *actrec = (ListenAction *) lfirst(p); - - switch (actrec->action) + foreach(p, pendingActions->actions) { - case LISTEN_LISTEN: - Exec_ListenCommit(actrec->channel); - break; - case LISTEN_UNLISTEN: - Exec_UnlistenCommit(actrec->channel); - break; - case LISTEN_UNLISTEN_ALL: - Exec_UnlistenAllCommit(); - break; + 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; + } } } @@ -1633,36 +1674,6 @@ AtAbort_Notify(void) ClearPendingActionsAndNotifies(); } -/* - * AtSubStart_Notify() --- Take care of subtransaction start. - * - * Push empty state for the new subtransaction. - */ -void -AtSubStart_Notify(void) -{ - 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); -} - /* * AtSubCommit_Notify() --- Take care of subtransaction commit. * @@ -1671,53 +1682,66 @@ AtSubStart_Notify(void) void AtSubCommit_Notify(void) { - 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); + int my_level = GetCurrentTransactionNestLevel(); - parentPendingNotifies = (NotificationList *) linitial(upperPendingNotifies); - upperPendingNotifies = list_delete_first(upperPendingNotifies); + /* 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; - Assert(list_length(upperPendingNotifies) == - GetCurrentTransactionNestLevel() - 2); + pendingActions = pendingActions->upper; - if (pendingNotifies == NULL) - { - /* easy, no notify events happened in current subxact */ - pendingNotifies = parentPendingNotifies; - } - else if (parentPendingNotifies == NULL) - { - /* easy, subxact's list becomes parent's */ + /* + * Mustn't try to eliminate duplicates here --- see queue_listen() + */ + pendingActions->actions = + list_concat(pendingActions->actions, + childPendingActions->actions); + pfree(childPendingActions); + } } - else + + /* If there are notifies at our nesting level, we must reparent them. */ + if (pendingNotifies != NULL && + pendingNotifies->nestingLevel >= my_level) { - /* - * 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; + Assert(pendingNotifies->nestingLevel == my_level); - pendingNotifies = parentPendingNotifies; - /* Insert all the subxact's events into parent, except for dups */ - foreach(l, childPendingNotifies->events) + if (pendingNotifies->upper == NULL || + pendingNotifies->upper->nestingLevel < my_level - 1) { - Notification *childn = (Notification *) lfirst(l); + /* 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); + if (!AsyncExistsPendingNotify(childn)) + AddEventToPendingNotifies(childn); + } + pfree(childPendingNotifies); } } } @@ -1733,23 +1757,31 @@ AtSubAbort_Notify(void) /* * 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. + * when CurTransactionContext is recycled. We still have to free the + * ActionList and NotificationList objects themselves, though, because + * those are allocated in TopTransactionContext. * - * 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. + * 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 (list_length(upperPendingActions) > my_level - 2) + while (pendingActions != NULL && + pendingActions->nestingLevel >= my_level) { - pendingActions = linitial_node(List, upperPendingActions); - upperPendingActions = list_delete_first(upperPendingActions); + ActionList *childPendingActions = pendingActions; + + pendingActions = pendingActions->upper; + pfree(childPendingActions); } - while (list_length(upperPendingNotifies) > my_level - 2) + while (pendingNotifies != NULL && + pendingNotifies->nestingLevel >= my_level) { - pendingNotifies = (NotificationList *) linitial(upperPendingNotifies); - upperPendingNotifies = list_delete_first(upperPendingNotifies); + NotificationList *childPendingNotifies = pendingNotifies; + + pendingNotifies = pendingNotifies->upper; + pfree(childPendingNotifies); } } @@ -2314,12 +2346,11 @@ static void ClearPendingActionsAndNotifies(void) { /* - * 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. + * Everything's allocated in either TopTransactionContext or the context + * for the subtransaction to which it corresponds. So, there's nothing + * to do here except rest the pointers; the space will be reclaimed when + * the contexts are deleted. */ - pendingActions = NIL; + pendingActions = NULL; pendingNotifies = NULL; } diff --git a/src/include/commands/async.h b/src/include/commands/async.h index c295dc67c6..90b682f64d 100644 --- a/src/include/commands/async.h +++ b/src/include/commands/async.h @@ -40,7 +40,6 @@ extern void Async_UnlistenAll(void); extern void PreCommit_Notify(void); extern void AtCommit_Notify(void); extern void AtAbort_Notify(void); -extern void AtSubStart_Notify(void); extern void AtSubCommit_Notify(void); extern void AtSubAbort_Notify(void); extern void AtPrepare_Notify(void); -- 2.17.2 (Apple Git-113)