From: Nathan Bossart <nathandbossart@gmail.com>
To: Melih Mutlu <m.melihmutlu@gmail.com>
Cc: Thomas Munro <thomas.munro@gmail.com>
Cc: Hayato Kuroda (Fujitsu) <kuroda.hayato@fujitsu.com>
Cc: pgsql-hackers@postgresql.org <pgsql-hackers@postgresql.org>
Subject: Re: wake up logical workers after ALTER SUBSCRIPTION
Date: Tue, 6 Dec 2022 13:29:54 -0800
Message-ID: <20221206212954.GA3403597@nathanxps13> (raw)
In-Reply-To: <20221206192551.GA3078082@nathanxps13>
References: <TYCPR01MB58703BC1BD6B1415CA4BAB7FF50F9@TYCPR01MB5870.jpnprd01.prod.outlook.com>
<20221127234528.GB878043@nathanxps13>
<20221130041028.GA1604177@nathanxps13>
<CA+hUKGJTQqnAAYAmdM9xkcs1e+5jOt34nvV622cY8cTmQcgVpA@mail.gmail.com>
<CA+hUKGJXO6LUn3wTf3s0Z8DvgbQROxPTnJDHCsQrhbr3pF3g8Q@mail.gmail.com>
<20221130050441.GA1677223@nathanxps13>
<20221202002130.GA2124877@nathanxps13>
<20221202192101.GB2277157@nathanxps13>
<CAGPVpCTUuKyxB5f3exgDMBRTivbN5uHVkD=YfXmgznrdnwVOQw@mail.gmail.com>
<20221206192551.GA3078082@nathanxps13>
On Tue, Dec 06, 2022 at 11:25:51AM -0800, Nathan Bossart wrote:
> On Tue, Dec 06, 2022 at 07:44:46PM +0300, Melih Mutlu wrote:
>> - When the state is SYNCDONE and the apply worker has to wake up to change
>> the state to READY.
>>
>> I think we already call logicalrep_worker_wakeup_ptr wherever it's needed
>> for the above cases? What am I missing here?
>
> IIUC we must restart all the apply workers for a subscription to enable
> two_phase mode. It looks like finish_sync_worker() only wakes up its own
> apply worker. I moved this logic to where the sync worker marks the state
> as SYNCDONE and added a check that two_phase mode is pending. Even so,
> there can still be unnecessary wakeups, but this adjustment should limit
> them.
Actually, that's not quite right. The sync worker will wake up the apply
worker to change the state from SYNCDONE to READY. AllTablesyncsReady()
checks that all tables are READY, so we need to wake up all the workers
when an apply worker changes the state to READY. Each worker will then
evaluate whether to restart for two_phase mode.
--
Nathan Bossart
Amazon Web Services: https://aws.amazon.comAttachments:
[text/x-diff] v7-0001-wake-up-logical-workers-as-needed-instead-of-rely.patch (6.4K, ../20221206212954.GA3403597@nathanxps13/2-v7-0001-wake-up-logical-workers-as-needed-instead-of-rely.patch)
download | inline diff:
From 1ca15608be05229b3349ba5840abceebb1497fe1 Mon Sep 17 00:00:00 2001
From: Nathan Bossart <nathandbossart@gmail.com>
Date: Mon, 21 Nov 2022 16:01:01 -0800
Subject: [PATCH v7 1/1] wake up logical workers as needed instead of relying
on periodic wakeups
---
src/backend/access/transam/xact.c | 3 ++
src/backend/commands/alter.c | 7 ++++
src/backend/commands/subscriptioncmds.c | 4 ++
src/backend/replication/logical/tablesync.c | 8 ++++
src/backend/replication/logical/worker.c | 46 +++++++++++++++++++++
src/include/replication/logicalworker.h | 3 ++
6 files changed, 71 insertions(+)
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.cindex 8086b857b9..dc00e66cfb 100644--- a/src/backend/access/transam/xact.c+++ b/src/backend/access/transam/xact.c@@ -47,6 +47,7 @@
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/snapbuild.h"
#include "replication/syncrep.h"
@@ -2360,6 +2361,7 @@ CommitTransaction(void)
AtEOXact_PgStat(true, is_parallel_worker);
AtEOXact_Snapshot(true, false);
AtEOXact_ApplyLauncher(true);
+ AtEOXact_LogicalRepWorkers(true);
pgstat_report_xact_timestamp(0);
CurrentResourceOwner = NULL;
@@ -2860,6 +2862,7 @@ AbortTransaction(void)
AtEOXact_HashTables(false);
AtEOXact_PgStat(false, is_parallel_worker);
AtEOXact_ApplyLauncher(false);
+ AtEOXact_LogicalRepWorkers(false);
pgstat_report_xact_timestamp(0);
}
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.cindex 10b6fe19a2..d095cd3ced 100644--- a/src/backend/commands/alter.c+++ b/src/backend/commands/alter.c@@ -59,6 +59,7 @@
#include "commands/user.h"
#include "miscadmin.h"
#include "parser/parse_func.h"
+#include "replication/logicalworker.h"
#include "rewrite/rewriteDefine.h"
#include "tcop/utility.h"
#include "utils/builtins.h"
@@ -279,6 +280,12 @@ AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name)
if (strncmp(new_name, "regress_", 8) != 0)
elog(WARNING, "subscriptions created by regression test cases should have names starting with \"regress_\"");
#endif
++ /*+ * Wake up the logical replication workers to handle this change+ * quickly.+ */+ LogicalRepWorkersWakeupAtCommit(objectId);
}
else if (nameCacheId >= 0)
{
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.cindex d673557ea4..d6993c26e5 100644--- a/src/backend/commands/subscriptioncmds.c+++ b/src/backend/commands/subscriptioncmds.c@@ -34,6 +34,7 @@
#include "nodes/makefuncs.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/slot.h"
#include "replication/walreceiver.h"
@@ -1362,6 +1363,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
InvokeObjectPostAlterHook(SubscriptionRelationId, subid, 0);
+ /* Wake up the logical replication workers to handle this change quickly. */+ LogicalRepWorkersWakeupAtCommit(subid);+
return myself;
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.cindex 94e813ac53..e253db371a 100644--- a/src/backend/replication/logical/tablesync.c+++ b/src/backend/replication/logical/tablesync.c@@ -105,6 +105,7 @@
#include "pgstat.h"
#include "replication/logicallauncher.h"
#include "replication/logicalrelation.h"
+#include "replication/logicalworker.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "replication/slot.h"
@@ -517,6 +518,13 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
rstate->relid, rstate->state,
rstate->lsn);
++ /*+ * We might be ready to enable two_phase mode. Wake up the+ * logical replication workers to handle this change quickly.+ */+ if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)+ LogicalRepWorkersWakeupAtCommit(MyLogicalRepWorker->subid);
}
}
else
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.cindex 96772e4d73..722f796c7a 100644--- a/src/backend/replication/logical/worker.c+++ b/src/backend/replication/logical/worker.c@@ -254,6 +254,8 @@ WalReceiverConn *LogRepWorkerWalRcvConn = NULL;
Subscription *MySubscription = NULL;
static bool MySubscriptionValid = false;
+static List *on_commit_wakeup_workers_subids = NIL;+
bool in_remote_transaction = false;
static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
@@ -4097,3 +4099,47 @@ reset_apply_error_context_info(void)
apply_error_callback_arg.remote_attnum = -1;
set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
}
++/*+ * Wakeup the stored subscriptions' workers on commit if requested.+ */+void+AtEOXact_LogicalRepWorkers(bool isCommit)+{+ if (isCommit && on_commit_wakeup_workers_subids != NIL)+ {+ ListCell *subid;++ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);+ foreach(subid, on_commit_wakeup_workers_subids)+ {+ List *workers;+ ListCell *worker;++ workers = logicalrep_workers_find(lfirst_oid(subid), true);+ foreach(worker, workers)+ logicalrep_worker_wakeup_ptr((LogicalRepWorker *) lfirst(worker));+ }+ LWLockRelease(LogicalRepWorkerLock);+ }++ on_commit_wakeup_workers_subids = NIL;+}++/*+ * Request wakeup of the workers for the given subscription ID on commit of the+ * transaction.+ *+ * This is used to ensure that the workers process assorted changes as soon as+ * possible.+ */+void+LogicalRepWorkersWakeupAtCommit(Oid subid)+{+ MemoryContext oldcxt;++ oldcxt = MemoryContextSwitchTo(TopTransactionContext);+ on_commit_wakeup_workers_subids = list_append_unique_oid(on_commit_wakeup_workers_subids,+ subid);+ MemoryContextSwitchTo(oldcxt);+}diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.hindex cd1b6e8afc..2c2340d758 100644--- a/src/include/replication/logicalworker.h+++ b/src/include/replication/logicalworker.h@@ -16,4 +16,7 @@ extern void ApplyWorkerMain(Datum main_arg);
extern bool IsLogicalWorker(void);
+extern void LogicalRepWorkersWakeupAtCommit(Oid subid);+extern void AtEOXact_LogicalRepWorkers(bool isCommit);+
#endif /* LOGICALWORKER_H */
--
2.25.1
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: nathandbossart@gmail.com, m.melihmutlu@gmail.com, thomas.munro@gmail.com, kuroda.hayato@fujitsu.com
Subject: Re: wake up logical workers after ALTER SUBSCRIPTION
In-Reply-To: <20221206212954.GA3403597@nathanxps13>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox