Only kill sync workers at commit time in subscription DDL

This allows a transaction abort to avoid killing those workers.

Author: Petr Jelinek <petr.jelinek@2ndquadrant.com>

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index b0aa69fe4b43de546f0b4dedae97c313f2f4812f..50c3c3b5e5e3763d795294cf545c1f1bdaa6f0ed 100644 (file)
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -2277,6 +2277,15 @@ PrepareTransaction(void)
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot PREPARE a transaction that has exported snapshots")));
 
+   /*
+    * Don't allow PREPARE but for transaction that has/might kill logical
+    * replication workers.
+    */
+   if (XactManipulatesLogicalReplicationWorkers())
+       ereport(ERROR,
+               (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+                errmsg("cannot PREPARE a transaction that has manipulated logical replication workers")));
+
    /* Prevent cancel/die interrupt while cleaning up */
    HOLD_INTERRUPTS();
 

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 6dc3f6ee0009dedec1b7116b2fc167a849a8eb5c..87824b8fec3fe11e53012e4035dda6aaafe80b2d 100644 (file)
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -597,7 +597,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
 
            RemoveSubscriptionRel(sub->oid, relid);
 
-           logicalrep_worker_stop(sub->oid, relid);
+           logicalrep_worker_stop_at_commit(sub->oid, relid);
 
            namespace = get_namespace_name(get_rel_namespace(relid));
            ereport(NOTICE,
@@ -819,6 +819,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
    char       *subname;
    char       *conninfo;
    char       *slotname;
+   List       *subworkers;
+   ListCell   *lc;
    char        originname[NAMEDATALEN];
    char       *err = NULL;
    RepOriginId originid;
@@ -909,15 +911,33 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 
    ReleaseSysCache(tup);
 
+   /*
+    * If we are dropping the replication slot, stop all the subscription
+    * workers immediately, so that the slot becomes accessible.  Otherwise
+    * just schedule the stopping for the end of the transaction.
+    *
+    * New workers won't be started because we hold an exclusive lock on the
+    * subscription till the end of the transaction.
+    */
+   LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+   subworkers = logicalrep_workers_find(subid, false);
+   LWLockRelease(LogicalRepWorkerLock);
+   foreach (lc, subworkers)
+   {
+       LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+       if (slotname)
+           logicalrep_worker_stop(w->subid, w->relid);
+       else
+           logicalrep_worker_stop_at_commit(w->subid, w->relid);
+   }
+   list_free(subworkers);
+
    /* Clean up dependencies */
    deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
 
    /* Remove any associated relation synchronization states. */
    RemoveSubscriptionRel(subid, InvalidOid);
 
-   /* Kill the apply worker so that the slot becomes accessible. */
-   logicalrep_worker_stop(subid, InvalidOid);
-
    /* Remove the origin tracking if exists. */
    snprintf(originname, sizeof(originname), "pg_%u", subid);
    originid = replorigin_by_name(originname, true);

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index d165d518e1b35d9630683641b2f1e67f27e3b932..0f9e5755b9e747b67f739e21664f6dbf7cea718b 100644 (file)
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -73,6 +73,14 @@ typedef struct LogicalRepCtxStruct
 
 LogicalRepCtxStruct *LogicalRepCtx;
 
+typedef struct LogicalRepWorkerId
+{
+   Oid subid;
+   Oid relid;
+} LogicalRepWorkerId;
+
+static List *on_commit_stop_workers = NIL;
+
 static void ApplyLauncherWakeup(void);
 static void logicalrep_launcher_onexit(int code, Datum arg);
 static void logicalrep_worker_onexit(int code, Datum arg);
@@ -249,6 +257,30 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
    return res;
 }
 
+/*
+ * Similar to logicalrep_worker_find(), but returns list of all workers for
+ * the subscription, instead just one.
+ */
+List *
+logicalrep_workers_find(Oid subid, bool only_running)
+{
+   int         i;
+   List       *res = NIL;
+
+   Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+   /* Search for attached worker for a given subscription id. */
+   for (i = 0; i < max_logical_replication_workers; i++)
+   {
+       LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+       if (w->in_use && w->subid == subid && (!only_running || w->proc))
+           res = lappend(res, w);
+   }
+
+   return res;
+}
+
 /*
  * Start new apply background worker.
  */
@@ -513,6 +545,27 @@ logicalrep_worker_stop(Oid subid, Oid relid)
    LWLockRelease(LogicalRepWorkerLock);
 }
 
+/*
+ * Request worker for specified sub/rel to be stopped on commit.
+ */
+void
+logicalrep_worker_stop_at_commit(Oid subid, Oid relid)
+{
+   LogicalRepWorkerId *wid;
+   MemoryContext       oldctx;
+
+   /* Make sure we store the info in context that survives until commit. */
+   oldctx = MemoryContextSwitchTo(TopTransactionContext);
+
+   wid = palloc(sizeof(LogicalRepWorkerId));
+   wid->subid = subid;
+   wid->relid = relid;
+
+   on_commit_stop_workers = lappend(on_commit_stop_workers, wid);
+
+   MemoryContextSwitchTo(oldctx);
+}
+
 /*
  * Wake up (using latch) any logical replication worker for specified sub/rel.
  */
@@ -753,15 +806,41 @@ ApplyLauncherShmemInit(void)
    }
 }
 
+/*
+ * Check whether current transaction has manipulated logical replication
+ * workers.
+ */
+bool
+XactManipulatesLogicalReplicationWorkers(void)
+{
+   return (on_commit_stop_workers != NIL);
+}
+
 /*
  * Wakeup the launcher on commit if requested.
  */
 void
 AtEOXact_ApplyLauncher(bool isCommit)
 {
-   if (isCommit && on_commit_launcher_wakeup)
-       ApplyLauncherWakeup();
+   if (isCommit)
+   {
+       ListCell *lc;
 
+       foreach (lc, on_commit_stop_workers)
+       {
+           LogicalRepWorkerId *wid = lfirst(lc);
+           logicalrep_worker_stop(wid->subid, wid->relid);
+       }
+
+       if (on_commit_launcher_wakeup)
+           ApplyLauncherWakeup();
+   }
+
+   /*
+    * No need to pfree on_commit_stop_workers.  It was allocated in
+    * transaction memory context, which is going to be cleaned soon.
+    */
+   on_commit_stop_workers = NIL;
    on_commit_launcher_wakeup = false;
 }
 

diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index aac7d326e22cf2d1cbe3dc9cca42f0384b7327b9..78016c448f34964507c23203bb57808010973edf 100644 (file)
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -22,6 +22,7 @@ extern Size ApplyLauncherShmemSize(void);
 extern void ApplyLauncherShmemInit(void);
 
 extern void ApplyLauncherWakeupAtCommit(void);
+extern bool XactManipulatesLogicalReplicationWorkers(void);
 extern void AtEOXact_ApplyLauncher(bool isCommit);
 
 extern bool IsLogicalLauncher(void);

diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 494a3a3d087d2dea9cccb5b2bfd3a0fbc40e5905..7b8728cced0bb1b0e7b4f1ff2abbf1a0d5f91f98 100644 (file)
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -71,9 +71,11 @@ extern bool in_remote_transaction;
 extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
                       bool only_running);
+extern List *logicalrep_workers_find(Oid subid, bool only_running);
 extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
                         Oid userid, Oid relid);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
+extern void logicalrep_worker_stop_at_commit(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);