Subject: Slow catchup of 2PC (twophase) transactions on replica in LR

From: Давыдов Виталий <[email-masked]>
Date: Thu, 22 Feb 2024 16:29:43 +0300

Dear All,
I'd like to present and talk about a problem when 2PC transactions are applied quite slowly on a replica during logical replication. There is a master and a replica with established logical replication from the master to the replica with twophase = true. With some load level on the master, the replica starts to lag behind the master, and the lag will be increasing. We have to significantly decrease the load on the master to allow replica to complete the catchup. Such problem may create significant difficulties in the production. The problem appears at least on REL_16_STABLE branch.
To reproduce the problem:
 * Setup logical replication from master to replica with subscription parameter twophase =  true. * Create some intermediate load on the master (use pgbench with custom sql with prepare+commit) * Optionally switch off the replica for some time (keep load on master). * Switch on the replica and wait until it reaches the master.
The replica will never reach the master with even some low load on the master. If to remove the load, the replica will reach the master for much greater time, than expected. I tried the same for regular transactions, but such problem doesn't appear even with a decent load.
I think, the main proplem of 2PC catchup bad performance - the lack of asynchronous commit support for 2PC. For regular transactions asynchronous commit is used on the replica by default (subscrition sycnronous_commit = off). It allows the replication worker process on the replica to avoid fsync (XLogFLush) and to utilize 100% CPU (the background wal writer or checkpointer will do fsync). I agree, 2PC are mostly used in multimaster configurations with two or more nodes which are performed synchronously, but when the node in catchup (node is not online in a multimaster cluster), asynchronous commit have to be used to speedup the catchup.
There is another thing that affects on the disbalance of the master and replica performance. When the master executes requestes from multiple clients, there is a fsync optimization takes place in XLogFlush. It allows to decrease the number of fsync in case when a number of parallel backends write to the WAL simultaneously. The replica applies received transactions in one thread sequentially, such optimization is not applied.
I see some possible solutions:
 * Implement asyncronous commit for 2PC transactions. * Do some hacking with enableFsync when it is possible.
I think, asynchronous commit support for 2PC transactions should significantly increase replica performance and help to solve this problem. I tried to implement it (like for usual transactions) but I've found another problem: 2PC state is stored in WAL on prepare, on commit we have to read 2PC state from WAL but the read is delayed until WAL is flushed by the background wal writer (read LSN should be less than flush LSN). Storing 2PC state in a shared memory (as it proposed earlier) may help.

I used the following query to monitor the catchup progress on the master:SELECT sent_lsn, pg_current_wal_lsn() FROM pg_stat_replication;
I used the following script for pgbench to the master:SELECT md5(random()::text) as mygid \gset
BEGIN;
DELETE FROM test WHERE v = pg_backend_pid();
INSERT INTO test(v) SELECT pg_backend_pid();
PREPARE TRANSACTION $$:mygid$$;
COMMIT PREPARED $$:mygid$$;
 
What do you think?
 
With best regards,
Vitaly Davydov

---

From: Amit Kapila <[email-masked]>
Date: Fri, 23 Feb 2024 08:53:49 +0530

On Thu, Feb 22, 2024 at 6:59 PM Давыдов Виталий
<[email-masked]> wrote:
> Setup logical replication from master to replica with subscription parameter twophase =  true.
> Create some intermediate load on the master (use pgbench with custom sql with prepare+commit)
> Optionally switch off the replica for some time (keep load on master).
> Switch on the replica and wait until it reaches the master.
>

I don't see we do anything specific for 2PC transactions to make them
behave differently than regular transactions with respect to
synchronous_commit setting. What makes you think so? Can you pin point
the code you are referring to?

> There is another thing that affects on the disbalance of the master and replica performance. When the master executes requestes from multiple clients, there is a fsync optimization takes place in X…

Right, I think for this we need to implement parallel apply.

> Implement asyncronous commit for 2PC transactions.
> Do some hacking with enableFsync when it is possible.
>

Can you be a bit more specific about what exactly you have in mind to
achieve the above solutions?

-- 
With Regards,
Amit Kapila.

---

From: Ajin Cherian <[email-masked]>
Date: Fri, 23 Feb 2024 15:52:11 +1100

On Fri, Feb 23, 2024 at 12:29 AM Давыдов Виталий <[email-masked]>
wrote:

> Dear All,
>
> I'd like to present and talk about a problem when 2PC transactions are
> applied quite slowly on a replica during logical replication. There is a
> master and a replica with established logical replication from the master
> to the replica with twophase = true. With some load level on the master,
> the replica starts to lag behind the master, and the lag will be
> increasing. We have to significantly decrease the load on the master to
> allow replica to complete the catchup. Such problem may create significant
> difficulties in the production. The problem appears at least on
> REL_16_STABLE branch.
>
>    - Setup logical replication from master to replica with subscription
>    parameter twophase =  true.
>    - Create some intermediate load on the master (use pgbench with custom
>    sql with prepare+commit)
>    - Optionally switch off the replica for some time (keep load on
>    master).
>    - Switch on the replica and wait until it reaches the master.
>
> The replica will never reach the master with even some low load on the
> master. If to remove the load, the replica will reach the master for much
> greater time, than expected. I tried the same for regular transactions, but
> such problem doesn't appear even with a decent load.
>
>
>
I tried this setup and I do see that the logical subscriber does reach the
master in a short time. I'm not sure what I'm missing. I stopped the
logical subscriber in between while pgbench was running and then started it
again and ran the following:
postgres=# SELECT sent_lsn, pg_current_wal_lsn() FROM pg_stat_replication;
 sent_lsn  | pg_current_wal_lsn
-----------+--------------------
 0/6793FA0 | 0/6793FA0 <=== caught up
(1 row)

My pgbench command:
pgbench postgres -p 6972 -c 2 -j 3 -f /home/ajin/test.sql -T 200 -P 5

my custom sql file:
cat test.sql
SELECT md5(random()::text) as mygid \gset
BEGIN;

regards,
Ajin Cherian
Fujitsu Australia

---

From: Давыдов Виталий <[email-masked]>
Date: Fri, 23 Feb 2024 20:11:46 +0300

Hi Amit,
Amit Kapila <[email-masked]> wrote:
I don't see we do anything specific for 2PC transactions to make them behave differently than regular transactions with respect to synchronous_commit setting. What makes you think so? Can you pin point the code you are referring to?Yes, sure. The function RecordTransactionCommitPrepared is called on prepared transaction commit (twophase.c). It calls XLogFlush unconditionally. The function RecordTransactionCommit (for regular transactions, xact.c) calls XLogFlush if synchronous_commit > OFF, otherwise it calls XLogSetAsyncXactLSN.

There is some comment in RecordTransactionCommitPrepared (by Bruce Momjian) that shows that async commit is not supported yet:
/*
* We don't currently try to sleep before flush here ... nor is there any
* support for async commit of a prepared xact (the very idea is probably
* a contradiction)
*/
/* Flush XLOG to disk */
XLogFlush(recptr);
Right, I think for this we need to implement parallel apply.Yes, parallel apply is a good point. But, I believe, it will not work if asynchronous commit is not supported. You have only one receiver process which should dispatch incoming messages to parallel workers. I guess, you will never reach such rate of parallel execution on replica as on the master with multiple backends.
 
Can you be a bit more specific about what exactly you have in mind to achieve the above solutions?My proposal is to implement async commit for 2PC transactions as it is for regular transactions. It should significantly speedup the catchup process. Then, think how to apply in parallel, which is much diffcult to do. The current problem is to get 2PC state from the WAL on commit prepared. At this moment, the WAL is not flushed yet, commit function waits until WAL with 2PC state is to be flushed. I just tried to do it in my sandbox and found such a problem. Inability to get 2PC state from unflushed WAL stops me right now. I think about possible solutions.

The idea with enableFsync is not a suitable solution, in general, I think. I just pointed it as an alternate idea. You just do enableFsync = false before prepare or commit prepared and do enableFsync = true after these functions. In this case, 2PC records will not be fsync-ed, but FlushPtr will be increased. Thus, 2PC state can be read from WAL on commit prepared without waiting. To make it work correctly, I guess, we have to do some additional work to keep more wal on the master and filter some duplicate transactions on the replica, if replica restarts during catchup.
 
With best regards,
 Vitaly Davydov

---

From: Amit Kapila <[email-masked]>
Date: Mon, 26 Feb 2024 18:54:55 +0530

On Fri, Feb 23, 2024 at 10:41 PM Давыдов Виталий
<[email-masked]> wrote:
> I don't see we do anything specific for 2PC transactions to make them behave differently than regular transactions with respect to synchronous_commit setting. What makes you think so? Can you pin point the code you are referring to?
>
> Yes, sure. The function RecordTransactionCommitPrepared is called on prepared transaction commit (twophase.c). It calls XLogFlush unconditionally. The function RecordTransactionCommit (for regular transactions, xact.c) calls XLogFlush if synchronous_commit > OFF, otherwise it calls XLogSetAsyncXactLSN.
>

It seems this comment is added in the commit 4a78cdeb where we added
async commit support. I think the reason is probably that when the WAL
record for prepared is already flushed then what will be the idea of
async commit here?

> Yes, parallel apply is a good point. But, I believe, it will not work if asynchronous commit is not supported. You have only one receiver process which should dispatch incoming messages to parallel workers. I guess, you will never reach such rate of parallel execution on replica as on the master with multiple backends.
>
>
> Can you be a bit more specific about what exactly you have in mind to achieve the above solutions?
>
> My proposal is to implement async commit for 2PC transactions as it is for regular transactions. It should significantly speedup the catchup process. Then, think how to apply in parallel, which is much diffcult to do. The current problem is to get 2PC state from the WAL on commit prepared. At this moment, the WAL is not flushed yet, commit function waits until WAL with 2PC state is to be flushed. I just tried to do it in my sandbox and found such a problem. Inability to get 2PC state from unflushed WAL stops me right now. I think about possible solutions.
>

At commit prepared, it seems we read prepare's WAL record, right? If
so, it is not clear to me do you see a problem with a flush of
commit_prepared or reading WAL for prepared or both of these.

-- 
With Regards,
Amit Kapila.

---

From: Давыдов Виталий <[email-masked]>
Date: Tue, 27 Feb 2024 14:19:41 +0300

Hi Amit,

Thank you for your interest in the discussion!

On Monday, February 26, 2024 16:24 MSK, Amit Kapila <[email-masked]> wrote:
 
I think the reason is probably that when the WAL record for prepared is already flushed then what will be the idea of async commit here?I think, the idea of async commit should be applied for both transactions: PREPARE and COMMIT PREPARED, which are actually two separate local transactions. For both these transactions we may call XLogSetAsyncXactLSN on commit instead of XLogFlush when async commit is enabled. When I use async commit, I mean to apply async commit to local transactions, not to a twophase (prepared) transaction itself.
 
At commit prepared, it seems we read prepare's WAL record, right? If so, it is not clear to me do you see a problem with a flush of commit_prepared or reading WAL for prepared or both of these.The problem with reading WAL is due to async commit of PREPARE TRANSACTION which saves 2PC in the WAL. At the moment of COMMIT PREPARED the WAL with PREPARE TRANSACTION 2PC state may not be XLogFlush-ed yet. So, PREPARE TRANSACTION should wait until its 2PC state is flushed.

I did some experiments with saving 2PC state in the local memory of logical replication worker and, I think, it worked and demonstrated much better performance. Logical replication worker utilized up to 100% CPU. I'm just concerned about possible problems with async commit for twophase transactions.

To be more specific, I've attached a patch to support async commit for twophase. It is not the final patch but it is presented only for discussion purposes. There were some attempts to save 2PC in memory in past but it was rejected. Now, there might be the second round to discuss it.

With best regards,
Vitaly

From 549f809fa122ca0842ec4bfc775afd08feee0d80 Mon Sep 17 00:00:00 2001
From: Vitaly Davydov <[email-masked]>
Date: Tue, 27 Feb 2024 14:02:23 +0300
Subject: [PATCH] Add asynchronous commit support for 2PC

---
 src/backend/access/transam/twophase.c | 111 +++++++++++++++++++++++++-
 1 file changed, 108 insertions(+), 3 deletions(-)

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index c6af8cfd7e..52f0853db8 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -109,6 +109,8 @@
 #include "utils/memutils.h"
 #include "utils/timestamp.h"
 
+#define POSTGRESQL_TWOPHASE_SUPPORT_ASYNC_COMMIT
+
 /*
  * Directory where Two-phase commit files reside within PGDATA
  */
@@ -163,6 +165,9 @@ typedef struct GlobalTransactionData
 	 */
 	XLogRecPtr	prepare_start_lsn;	/* XLOG offset of prepare record start */
 	XLogRecPtr	prepare_end_lsn;	/* XLOG offset of prepare record end */
+	void*       prepare_2pc_mem_data;
+	size_t      prepare_2pc_mem_len;
+	pid_t       prepare_2pc_proc;
 	TransactionId xid;			/* The GXACT id */
 
 	Oid			owner;			/* ID of user that executed the xact */
@@ -427,6 +432,9 @@ MarkAsPreparing(TransactionId xid, const char *gid,
 
 	MarkAsPreparingGuts(gxact, xid, gid, prepared_at, owner, databaseid);
 
+	Assert(gxact->prepare_2pc_mem_data == NULL);
+	Assert(gxact->prepare_2pc_proc == 0);
+
 	gxact->ondisk = false;
 
 	/* And insert it into the active array */
@@ -1129,6 +1137,8 @@ StartPrepare(GlobalTransaction gxact)
 	}
 }
 
+extern bool IsLogicalWorker(void);
+
 /*
  * Finish preparing state data and writing it to WAL.
  */
@@ -1167,6 +1177,37 @@ EndPrepare(GlobalTransaction gxact)
 				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
 				 errmsg("two-phase state file maximum length exceeded")));
 
+	Assert(gxact->prepare_2pc_mem_data == NULL);
+	Assert(gxact->prepare_2pc_proc == 0);
+
+	if (IsLogicalWorker())
+	{
+		size_t			len = 0;
+		size_t			offset = 0;
+
+		for (record = records.head; record != NULL; record = record->next)
+			len += record->len;
+
+		if (len > 0)
+		{
+			MemoryContext	oldmemctx;
+
+			oldmemctx = MemoryContextSwitchTo(TopMemoryContext);
+
+			gxact->prepare_2pc_mem_data = palloc(len);
+			gxact->prepare_2pc_mem_len = len;
+			gxact->prepare_2pc_proc = getpid();
+
+			for (record = records.head; record != NULL; record = record->next)
+			{
+				memcpy((char *)gxact->prepare_2pc_mem_data + offset, record->data, record->len);
+				offset += record->len;
+			}
+
+			MemoryContextSwitchTo(oldmemctx);
+		}
+	}
+
 	/*
 	 * Now writing 2PC state data to WAL. We let the WAL's CRC protection
 	 * cover us, so no need to calculate a separate CRC.
@@ -1202,8 +1243,24 @@ EndPrepare(GlobalTransaction gxact)
 								   gxact->prepare_end_lsn);
 	}
 
+#if !defined(POSTGRESQL_TWOPHASE_SUPPORT_ASYNC_COMMIT)
+
 	XLogFlush(gxact->prepare_end_lsn);
 
+#else
+
+	if (synchronous_commit > SYNCHRONOUS_COMMIT_OFF)
+	{
+		/* Flush XLOG to disk */
+		XLogFlush(gxact->prepare_end_lsn);
+	}
+	else
+	{
+		XLogSetAsyncXactLSN(gxact->prepare_end_lsn);
+	}
+
+#endif
+
 	/* If we crash now, we have prepared: WAL replay will fix things */
 
 	/* Store record's start location to read that later on Commit */
@@ -1495,6 +1552,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	xl_xact_stats_item *commitstats;
 	xl_xact_stats_item *abortstats;
 	SharedInvalidationMessage *invalmsgs;
+	bool		is_local_2pc_buf = false;
 
 	/*
 	 * Validate the GID, and lock the GXACT to ensure that two backends do not
@@ -1509,12 +1567,19 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	 * in WAL files if the LSN is after the last checkpoint record, or moved
 	 * to disk if for some reason they have lived for a long time.
 	 */
-	if (gxact->ondisk)
+	if (gxact->prepare_2pc_mem_data != NULL && gxact->prepare_2pc_proc == getpid())
+	{
+		Assert(IsLogicalWorker());
+		Assert(gxact->prepare_2pc_proc == getpid());
+		buf = gxact->prepare_2pc_mem_data;
+		/* ereport(LOG, errmsg("%s:%d", __FUNCTION__, __LINE__)); */
+		is_local_2pc_buf = true;
+	}
+	else if (gxact->ondisk)
 		buf = ReadTwoPhaseFile(xid, false);
 	else
 		XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
 
-
 	/*
 	 * Disassemble the header area
 	 */
@@ -1638,6 +1703,19 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	/* Clear shared memory state */
 	RemoveGXact(gxact);
 
+	if (gxact->prepare_2pc_mem_data != NULL)
+	{
+		MemoryContext		memctx;
+		//Assert(gxact->prepare_2pc_proc == getpid());
+		memctx = MemoryContextSwitchTo(TopMemoryContext);
+		if (gxact->prepare_2pc_proc == getpid())
+			pfree(gxact->prepare_2pc_mem_data);
+		gxact->prepare_2pc_mem_data = NULL;
+		gxact->prepare_2pc_mem_len = 0;
+		gxact->prepare_2pc_proc = 0;
+		MemoryContextSwitchTo(memctx);
+	}
+
 	/*
 	 * Release the lock as all callbacks are called and shared memory cleanup
 	 * is done.
@@ -1657,7 +1735,8 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 
 	RESUME_INTERRUPTS();
 
-	pfree(buf);
+	if (!is_local_2pc_buf)
+		pfree(buf);
 }
 
 /*
@@ -2349,12 +2428,38 @@ RecordTransactionCommitPrepared(TransactionId xid,
 	 * a contradiction)
 	 */
 
+#if !defined(POSTGRESQL_TWOPHASE_SUPPORT_ASYNC_COMMIT)
+
 	/* Flush XLOG to disk */
 	XLogFlush(recptr);
 
 	/* Mark the transaction committed in pg_xact */
 	TransactionIdCommitTree(xid, nchildren, children);
 
+#else
+
+	if (synchronous_commit > SYNCHRONOUS_COMMIT_OFF)
+	{
+		/* Flush XLOG to disk */
+		XLogFlush(recptr);
+
+		/* Mark the transaction committed in pg_xact */
+		TransactionIdCommitTree(xid, nchildren, children);
+	}
+	else
+	{
+		XLogSetAsyncXactLSN(recptr);
+
+		/*
+		 * We must not immediately update the CLOG, since we didn't flush the
+		 * XLOG. Instead, we store the LSN up to which the XLOG must be
+		 * flushed before the CLOG may be updated.
+		 */
+		TransactionIdAsyncCommitTree(xid, nchildren, children, recptr);
+	}
+
+#endif
+
 	/* Checkpoint can proceed now */
 	MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
 
-- 
2.34.1

---

From: Amit Kapila <[email-masked]>
Date: Tue, 27 Feb 2024 18:30:28 +0530

On Tue, Feb 27, 2024 at 4:49 PM Давыдов Виталий
<[email-masked]> wrote:
> I think the reason is probably that when the WAL record for prepared is already flushed then what will be the idea of async commit here?
>
> I think, the idea of async commit should be applied for both transactions: PREPARE and COMMIT PREPARED, which are actually two separate local transactions. For both these transactions we may call XLogSetAsyncXactLSN on commit instead of XLogFlush when async commit is enabled. When I use async commit, I mean to apply async commit to local transactions, not to a twophase (prepared) transaction itself.
>
>
> At commit prepared, it seems we read prepare's WAL record, right? If so, it is not clear to me do you see a problem with a flush of commit_prepared or reading WAL for prepared or both of these.
>
> The problem with reading WAL is due to async commit of PREPARE TRANSACTION which saves 2PC in the WAL. At the moment of COMMIT PREPARED the WAL with PREPARE TRANSACTION 2PC state may not be XLogFlush-ed yet.
>

As we do XLogFlush() at the time of prepare then why it is not
available? OR are you talking about this state after your idea/patch
where you are trying to make both Prepare and Commit_prepared records
async?

 So, PREPARE TRANSACTION should wait until its 2PC state is flushed.
> To be more specific, I've attached a patch to support async commit for twophase. It is not the final patch but it is presented only for discussion purposes. There were some attempts to save 2PC in memory in past but it was rejected.
>

It would be good if you could link those threads.

-- 
With Regards,
Amit Kapila.

---

From: Давыдов Виталий <[email-masked]>
Date: Tue, 27 Feb 2024 16:34:47 +0300

Hi Amit,

On Tuesday, February 27, 2024 16:00 MSK, Amit Kapila <[email-masked]> wrote:
As we do XLogFlush() at the time of prepare then why it is not available? OR are you talking about this state after your idea/patch where you are trying to make both Prepare and Commit_prepared records async?Right, I'm talking about my patch where async commit is implemented. There is no such problem with reading 2PC from not flushed WAL in the vanilla because XLogFlush is called unconditionally, as you've described. But an attempt to add some async stuff leads to the problem of reading not flushed WAL. It is why I store 2pc state in the local memory in my patch.
It would be good if you could link those threads.Sure, I will find and add some links to the discussions from past.

Thank you!

With best regards,
Vitaly

--
With Regards,
Amit Kapila.

---

From: Давыдов Виталий <[email-masked]>
Date: Thu, 29 Feb 2024 20:34:42 +0300

Dear All,

Consider, please, my patch for async commit for twophase transactions. It can be applicable when catchup performance is not enought with publication parameter twophase = on.

The key changes are:
 * Use XLogSetAsyncXactLSN instead of XLogFlush as it is for usual transactions. * In case of async commit only, save 2PC state in the pg_twophase file (but not fsync it) in addition to saving in the WAL. The file is used as an alternative to storing 2pc state in the memory. * On recovery, reject pg_twophase files with future xids.Probably, 2PC async commit should be enabled by a GUC (not implemented in the patch).

With best regards,
Vitaly

From cbaaa7270d771f9ccd6def08f0f02ce61dc15ff6 Mon Sep 17 00:00:00 2001
From: Vitaly Davydov <[email-masked]>
Date: Thu, 29 Feb 2024 18:58:13 +0300
Subject: [PATCH] Async commit support for twophase transactions

---
 src/backend/access/transam/twophase.c | 171 +++++++++++++++++++++-----
 1 file changed, 138 insertions(+), 33 deletions(-)

index 234c8d08eb..352266be14 100644
 
+#define POSTGRESQL_TWOPHASE_SUPPORT_ASYNC_COMMIT
+
 /*
  * Directory where Two-phase commit files reside within PGDATA
  */
@@ -169,6 +171,7 @@ typedef struct GlobalTransactionData
 	BackendId	locking_backend;	/* backend currently working on the xact */
 	bool		valid;			/* true if PGPROC entry is in proc array */
 	bool		ondisk;			/* true if prepare state file is on disk */
+	bool		infile;			/* true if prepared state saved in file (but not fsync-ed) */
 	bool		inredo;			/* true if entry was added via xlog_redo */
 	char		gid[GIDSIZE];	/* The GID assigned to the prepared xact */
 }			GlobalTransactionData;
@@ -227,12 +230,14 @@ static void RemoveGXact(GlobalTransaction gxact);
 static void XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len);
 static char *ProcessTwoPhaseBuffer(TransactionId xid,
 								   XLogRecPtr prepare_start_lsn,
-								   bool fromdisk, bool setParent, bool setNextXid);
+								   bool fromdisk, bool setParent, bool setNextXid,
+								   const char *filename);
 static void MarkAsPreparingGuts(GlobalTransaction gxact, TransactionId xid,
 								const char *gid, TimestampTz prepared_at, Oid owner,
 								Oid databaseid);
 static void RemoveTwoPhaseFile(TransactionId xid, bool giveWarning);
-static void RecreateTwoPhaseFile(TransactionId xid, void *content, int len);
+static void RemoveTwoPhaseFileByName(const char *filename, bool giveWarning);
+static void RecreateTwoPhaseFile(TransactionId xid, void *content, int len, bool dosync);
 
 /*
  * Initialization of shared memory
@@ -427,6 +432,7 @@ MarkAsPreparing(TransactionId xid, const char *gid,

+	gxact->infile = false;
 	gxact->ondisk = false;
 
 	/* And insert it into the active array */
@@ -1204,6 +1210,37 @@ EndPrepare(GlobalTransaction gxact)
 				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
 				 errmsg("two-phase state file maximum length exceeded")));
 
+#ifdef POSTGRESQL_TWOPHASE_SUPPORT_ASYNC_COMMIT
+
+	Assert(gxact->infile == false);
+
+	if (synchronous_commit == SYNCHRONOUS_COMMIT_OFF)
+	{
+		char		   *buf;
+		size_t			len = 0;
+		size_t			offset = 0;
+
+		for (record = records.head; record != NULL; record = record->next)
+			len += record->len;
+
+		if (len > 0)
+		{
+			buf = palloc(len);
+
+			for (record = records.head; record != NULL; record = record->next)
+			{
+				memcpy(buf + offset, record->data, record->len);
+				offset += record->len;
+			}
+
+			RecreateTwoPhaseFile(gxact->xid, buf, len, false);
+			pfree(buf);
+			gxact->infile = true;
+		}
+	}
+
+#endif
+
 	/*
 	 * Now writing 2PC state data to WAL. We let the WAL's CRC protection
 	 * cover us, so no need to calculate a separate CRC.
@@ -1239,8 +1276,24 @@ EndPrepare(GlobalTransaction gxact)
 								   gxact->prepare_end_lsn);
 	}

+#else
+
 
 	/* Store record's start location to read that later on Commit */
@@ -1546,12 +1599,11 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	 * in WAL files if the LSN is after the last checkpoint record, or moved
 	 * to disk if for some reason they have lived for a long time.
 	 */
-	if (gxact->ondisk)
+	if (gxact->infile || gxact->ondisk)
 		buf = ReadTwoPhaseFile(xid, false);
 	else
 		XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
 
-
 	/*
 	 * Disassemble the header area
 	 */
@@ -1687,7 +1739,7 @@ FinishPreparedTransaction(const char *gid, bool isCommit)
 	/*
 	 * And now we can clean up any files we may have left.
 	 */
-	if (gxact->ondisk)
+	if (gxact->infile || gxact->ondisk)
 		RemoveTwoPhaseFile(xid, true);
 
 	MyLockedGxact = NULL;
@@ -1741,6 +1793,20 @@ RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
 					 errmsg("could not remove file \"%s\": %m", path)));
 }
 
+static void
+RemoveTwoPhaseFileByName(const char *filename, bool giveWarning)
+{
+	char		path[MAXPGPATH];
+
+	snprintf(path, MAXPGPATH, TWOPHASE_DIR "/%s", filename);
+
+	if (unlink(path))
+		if (errno != ENOENT || giveWarning)
+			ereport(WARNING,
+					(errcode_for_file_access(),
+					 errmsg("could not remove file \"%s\": %m", path)));
+}
+
 /*
  * Recreates a state file. This is used in WAL replay and during
  * checkpoint creation.
@@ -1748,7 +1814,7 @@ RemoveTwoPhaseFile(TransactionId xid, bool giveWarning)
  * Note: content and len don't include CRC.
  */
 static void
-RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
+RecreateTwoPhaseFile(TransactionId xid, void *content, int len, bool dosync)
 {
 	char		path[MAXPGPATH];
 	pg_crc32c	statefile_crc;
@@ -1791,16 +1857,19 @@ RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
 	}
 	pgstat_report_wait_end();
 
-	/*
-	 * We must fsync the file because the end-of-replay checkpoint will not do
-	 * so, there being no GXACT in shared memory yet to tell it to.
-	 */
-	pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_SYNC);
-	if (pg_fsync(fd) != 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not fsync file \"%s\": %m", path)));
-	pgstat_report_wait_end();
+	if (dosync)
+	{
+		/*
+		* We must fsync the file because the end-of-replay checkpoint will not do
+		* so, there being no GXACT in shared memory yet to tell it to.
+		*/
+		pgstat_report_wait_start(WAIT_EVENT_TWOPHASE_FILE_SYNC);
+		if (pg_fsync(fd) != 0)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					errmsg("could not fsync file \"%s\": %m", path)));
+		pgstat_report_wait_end();
+	}
 
 	if (CloseTransientFile(fd) != 0)
 		ereport(ERROR,
@@ -1871,7 +1940,8 @@ CheckPointTwoPhase(XLogRecPtr redo_horizon)
 			int			len;
 
 			XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, &len);
-			RecreateTwoPhaseFile(gxact->xid, buf, len);
+			RecreateTwoPhaseFile(gxact->xid, buf, len, true);
+			gxact->infile = true;
 			gxact->ondisk = true;
 			gxact->prepare_start_lsn = InvalidXLogRecPtr;
 			gxact->prepare_end_lsn = InvalidXLogRecPtr;
@@ -1930,7 +2000,8 @@ restoreTwoPhaseData(void)
 			xid = XidFromFullTransactionId(fxid);
 
 			buf = ProcessTwoPhaseBuffer(xid, InvalidXLogRecPtr,
-										true, false, false);
+										true, false, false,
+										clde->d_name);
 			if (buf == NULL)
 				continue;
 
@@ -1997,7 +2068,8 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
 
 		buf = ProcessTwoPhaseBuffer(xid,
 									gxact->prepare_start_lsn,
-									gxact->ondisk, false, true);
+									gxact->ondisk, false, true,
+									NULL);
 
 		if (buf == NULL)
 			continue;
@@ -2072,7 +2144,8 @@ StandbyRecoverPreparedTransactions(void)
 
 		buf = ProcessTwoPhaseBuffer(xid,
 									gxact->prepare_start_lsn,
-									gxact->ondisk, false, false);
+									gxact->ondisk, false, false,
+									NULL);
 		if (buf != NULL)
 			pfree(buf);
 	}
@@ -2124,7 +2197,8 @@ RecoverPreparedTransactions(void)
 		 */
 		buf = ProcessTwoPhaseBuffer(xid,
 									gxact->prepare_start_lsn,
-									gxact->ondisk, true, false);
+									gxact->ondisk, true, false,
+									NULL);
 		if (buf == NULL)
 			continue;
 
@@ -2202,7 +2276,8 @@ static char *
 ProcessTwoPhaseBuffer(TransactionId xid,
 					  XLogRecPtr prepare_start_lsn,
 					  bool fromdisk,
-					  bool setParent, bool setNextXid)
+					  bool setParent, bool setNextXid,
+					  const char *filename)
 {
 	FullTransactionId nextXid = TransamVariables->nextXid;
 	TransactionId origNextXid = XidFromFullTransactionId(nextXid);
@@ -2216,40 +2291,43 @@ ProcessTwoPhaseBuffer(TransactionId xid,
 	if (!fromdisk)
 		Assert(prepare_start_lsn != InvalidXLogRecPtr);
 
-	/* Already processed? */
-	if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
+	/* Reject XID if too new */
+	if (TransactionIdFollowsOrEquals(xid, origNextXid))
 	{
 		if (fromdisk)
 		{
 			ereport(WARNING,
-					(errmsg("removing stale two-phase state file for transaction %u",
+					(errmsg("removing future two-phase state file for transaction %u",
 							xid)));
-			RemoveTwoPhaseFile(xid, true);
+			if (filename)
+				RemoveTwoPhaseFileByName(filename, true);
+			else
+				RemoveTwoPhaseFile(xid, true);
 		}
 		else
 		{
 			ereport(WARNING,
-					(errmsg("removing stale two-phase state from memory for transaction %u",
+					(errmsg("removing future two-phase state from memory for transaction %u",
 							xid)));
 			PrepareRedoRemove(xid, true);
 		}
 		return NULL;
 	}
 
-	/* Reject XID if too new */
-	if (TransactionIdFollowsOrEquals(xid, origNextXid))
+	/* Already processed? */
+	if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
 	{
 		if (fromdisk)
 		{
 			ereport(WARNING,
-					(errmsg("removing future two-phase state file for transaction %u",
+					(errmsg("removing stale two-phase state file for transaction %u",
 							xid)));
 			RemoveTwoPhaseFile(xid, true);
 		}
 		else
 		{
 			ereport(WARNING,
-					(errmsg("removing future two-phase state from memory for transaction %u",
+					(errmsg("removing stale two-phase state from memory for transaction %u",
 							xid)));
 			PrepareRedoRemove(xid, true);
 		}
@@ -2388,12 +2466,38 @@ RecordTransactionCommitPrepared(TransactionId xid,
 	 * a contradiction)
 	 */

 
+#else
+
 
@@ -2567,6 +2671,7 @@ PrepareRedoAdd(char *buf, XLogRecPtr start_lsn,
 	gxact->locking_backend = InvalidBackendId;
 	gxact->valid = false;
 	gxact->ondisk = XLogRecPtrIsInvalid(start_lsn);
+	gxact->infile = gxact->ondisk;
 	gxact->inredo = true;		/* yes, added in redo */
 	strcpy(gxact->gid, gid);
 
@@ -2625,7 +2730,7 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning)
 	 * And now we can clean up any files we may have left.
 	 */
 	elog(DEBUG2, "removing 2PC data for transaction %u", xid);
-	if (gxact->ondisk)
+	if (gxact->infile || gxact->ondisk)
 		RemoveTwoPhaseFile(xid, giveWarning);
 	RemoveGXact(gxact);
 }
@@ -2673,7 +2778,7 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn,
 			 * do this optimization if we encounter many collisions in GID
 			 * between publisher and subscriber.
 			 */
-			if (gxact->ondisk)
+			if (gxact->infile || gxact->ondisk)
 				buf = ReadTwoPhaseFile(gxact->xid, false);
 			else
 			{
-- 
2.34.1

---

From: Heikki Linnakangas <[email-masked]>
Date: Tue, 5 Mar 2024 11:05:40 +0200

On 29/02/2024 19:34, Давыдов Виталий wrote:
> Dear All,
> 
> Consider, please, my patch for async commit for twophase transactions. 
> It can be applicable when catchup performance is not enought with 
> publication parameter twophase = on.
> 
> The key changes are:
> 
>   * Use XLogSetAsyncXactLSN instead of XLogFlush as it is for usual
>     transactions.
>   * In case of async commit only, save 2PC state in the pg_twophase file
>     (but not fsync it) in addition to saving in the WAL. The file is
>     used as an alternative to storing 2pc state in the memory.
>   * On recovery, reject pg_twophase files with future xids.
> 
> Probably, 2PC async commit should be enabled by a GUC (not implemented 
> in the patch).

In a nutshell, this changes PREPARE TRANSACTION so that if 
synchronous_commit is 'off', the PREPARE TRANSACTION is not fsync'd to 
disk. So if you crash after the PREPARE TRANSACTION has returned, the 
transaction might be lost. I think that's completely unacceptable.

If you're ok to lose the prepared state of twophase transactions on 
crash, why don't you create the subscription with 'two_phase=off' to 
begin with?

-- 
Heikki Linnakangas
Neon (https://neon.tech)

---

From: Давыдов Виталий <[email-masked]>
Date: Tue, 05 Mar 2024 17:29:32 +0300

Hi Heikki,

Thank you for the reply.

On Tuesday, March 05, 2024 12:05 MSK, Heikki Linnakangas <[email-masked]> wrote:
 In a nutshell, this changes PREPARE TRANSACTION so that if
synchronous_commit is 'off', the PREPARE TRANSACTION is not fsync'd to
disk. So if you crash after the PREPARE TRANSACTION has returned, the
transaction might be lost. I think that's completely unacceptable. 
You are right, the prepared transaction might be lost after crash. The same may happen with regular transactions that are not fsync-ed on replica in logical replication by default. The subscription parameter synchronous_commit is OFF by default. I'm not sure, is there some auto recovery for regular transactions? I think, the main difference between these two cases - how to manually recover when some PREPARE TRANSACTION or COMMIT PREPARED are lost. For regular transactions, some updates or deletes in tables on replica may be enough to fix the problem. For twophase transactions, it may be harder to fix it by hands, but it is possible, I believe. If you create a custom solution that is based on twophase transactions (like multimaster) such auto recovery may happen automatically. Another solution is to ignore errors on commit prepared if the corresponding prepared tx is missing. I don't know other risks that may happen with async commit of twophase transactions.
 If you're ok to lose the prepared state of twophase transactions on
crash, why don't you create the subscription with 'two_phase=off' to
begin with?In usual work, the subscription has two_phase = on. I have to change this option at catchup stage only, but this parameter can not be altered. There was a patch proposal in past to implement altering of two_phase option, but it was rejected. I think, the recreation of the subscription with two_phase = off will not work.

I believe, async commit for twophase transactions on catchup will significantly improve the catchup performance. It is worth to think about such feature.

P.S. We might introduce a GUC option to allow async commit for twophase transactions. By default, sync commit will be applied for twophase transactions, as it is now.

With best regards,
Vitaly Davydov

---

From: Amit Kapila <[email-masked]>
Date: Wed, 6 Mar 2024 17:55:54 +0530

On Tue, Mar 5, 2024 at 7:59 PM Давыдов Виталий <[email-masked]> wrote:
> You are right, the prepared transaction might be lost after crash. The same may happen with regular transactions that are not fsync-ed on replica in logical replication by default. The subscription parameter synchronous_commit is OFF by default. I'm not sure, is there some auto recovery for regular transactions?
>

Unless the commit WAL is not flushed, we wouldn't have updated the
replication origin's LSN and neither the walsender would increase the
confirmed_flush_lsn for the corresponding slot till the commit is
flushed on subscriber. So, if the subscriber crashed before flushing
the commit record, server should send the same transaction again. The
same should be true for prepared transaction stuff as well.

-- 
With Regards,
Amit Kapila.

---

From: Ajin Cherian <[email-masked]>
Date: Thu, 4 Apr 2024 16:23:15 +1100

On Wed, Mar 6, 2024 at 1:29 AM Давыдов Виталий <[email-masked]>
wrote:

> In usual work, the subscription has two_phase = on. I have to change this
> option at catchup stage only, but this parameter can not be altered. There
> was a patch proposal in past to implement altering of two_phase option, but
> it was rejected. I think, the recreation of the subscription with two_phase
> = off will not work.
>
>
>
The altering of two_phase was restricted because if there was a previously
prepared transaction on the subscriber when the two_phase was on, and then
it was turned off, the apply worker on the subscriber would re-apply the
transaction a second time and this might result in an inconsistent replica.
Here's a patch that allows toggling two_phase option provided that there
are no pending uncommitted prepared transactions on the subscriber for that
subscription.

Thanks to Kuroda-san for working on the patch.

regards,
Ajin Cherian
Fujitsu Australia

---

From: Amit Kapila <[email-masked]>
Date: Thu, 4 Apr 2024 11:07:49 +0530

On Thu, Apr 4, 2024 at 10:53 AM Ajin Cherian <[email-masked]> wrote:
>
> On Wed, Mar 6, 2024 at 1:29 AM Давыдов Виталий <[email-masked]> wrote:
>>
>> In usual work, the subscription has two_phase = on. I have to change this option at catchup stage only, but this parameter can not be altered. There was a patch proposal in past to implement altering of two_phase option, but it was rejected. I think, the recreation of the subscription with two_phase = off will not work.
>>
>>
>
> The altering of two_phase was restricted because if there was a previously prepared transaction on the subscriber when the two_phase was on, and then it was turned off, the apply worker on the subscriber would re-apply the transaction a second time and this might result in an inconsistent replica.
> Here's a patch that allows toggling two_phase option provided that there are no pending uncommitted prepared transactions on the subscriber for that subscription.
>

I think this would probably be better than the current situation but
can we think of a solution to allow toggling the value of two_phase
even when prepared transactions are present? Can you please summarize
the reason for the problems in doing that and the solutions, if any?

-- 
With Regards,
Amit Kapila.