public inbox for [email protected]
help / color / mirror / Atom feedFrom: Amit Kapila <[email protected]>
To: [email protected] <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
Date: Thu, 1 Dec 2022 18:09:54 +0530
Message-ID: <CAA4eK1Jwh7j86Egk1cye=x2R_yrTjzXGj7Fx12wVybBAEq91kA@mail.gmail.com> (raw)
In-Reply-To: <CAA4eK1LGKYUDFZ_jFPrU497wQf2HNvt5a+tCTpqSeWSG6kfpSA@mail.gmail.com>
References: <CAA4eK1JQTDXTfvJ5d+L0ggG4+doyd0Xji=e0OJsb=qcn_jWALA@mail.gmail.com>
<OS0PR01MB5716D6D2765E54DC739F288E940A9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716997A7115715F9E4EE520940D9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1JEFVZcymteMYXhYghiX7Lb=MKuEsfxgD1VfaRrLCHyzg@mail.gmail.com>
<OS0PR01MB571663F65904D00895AD159994109@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1JuRPUY2Dx=vUToRksTmW0ptqMVT3K32g0368f_ZCk-zg@mail.gmail.com>
<OS0PR01MB5716B802A1733548A99761AE94129@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1LGKYUDFZ_jFPrU497wQf2HNvt5a+tCTpqSeWSG6kfpSA@mail.gmail.com>
On Wed, Nov 30, 2022 at 4:23 PM Amit Kapila <[email protected]> wrote:
>
> 2.
> + /*
> + * The stream lock is released when processing changes in a
> + * streaming block, so the leader needs to acquire the lock here
> + * before entering PARTIAL_SERIALIZE mode to ensure that the
> + * parallel apply worker will wait for the leader to release the
> + * stream lock.
> + */
> + if (in_streamed_transaction &&
> + action != LOGICAL_REP_MSG_STREAM_STOP)
> + {
> + pa_lock_stream(winfo->shared->xid, AccessExclusiveLock);
>
> This comment is not completely correct because we can even acquire the
> lock for the very streaming chunk. This check will work but doesn't
> appear future-proof or at least not very easy to understand though I
> don't have a better suggestion at this stage. Can we think of a better
> check here?
>
One idea is that we acquire this lock every time and callers like
stream_commit are responsible to release it. Also, we can handle the
close of stream file in the respective callers. I think that will make
this part of the patch easier to follow.
Some other comments:
=====================
1. The handling of buffile inside pa_stream_abort() looks bit ugly to
me. I think you primarily required it because the buffile opened by
parallel apply worker is in CurrentResourceOwner. Can we think of
having a new resource owner to apply spooled messages? I think that
will avoid the need to have a special purpose code to handle buffiles
in parallel apply worker.
2.
@@ -564,6 +571,7 @@ handle_streamed_transaction(LogicalRepMsgType
action, StringInfo s)
TransactionId current_xid;
ParallelApplyWorkerInfo *winfo;
TransApplyAction apply_action;
+ StringInfoData original_msg;
apply_action = get_transaction_apply_action(stream_xid, &winfo);
@@ -573,6 +581,8 @@ handle_streamed_transaction(LogicalRepMsgType
action, StringInfo s)
Assert(TransactionIdIsValid(stream_xid));
+ original_msg = *s;
+
/*
* We should have received XID of the subxact as the first part of the
* message, so extract it.
@@ -596,10 +606,14 @@ handle_streamed_transaction(LogicalRepMsgType
action, StringInfo s)
stream_write_change(action, s);
return true;
+ case TRANS_LEADER_PARTIAL_SERIALIZE:
case TRANS_LEADER_SEND_TO_PARALLEL:
Assert(winfo);
- pa_send_data(winfo, s->len, s->data);
+ if (apply_action == TRANS_LEADER_SEND_TO_PARALLEL)
+ pa_send_data(winfo, s->len, s->data);
+ else
+ stream_write_change(action, &original_msg);
Please add the comment to specify the reason to remember the original string.
3.
@@ -1797,8 +1907,8 @@ apply_spooled_messages(TransactionId xid, XLogRecPtr lsn)
changes_filename(path, MyLogicalRepWorker->subid, xid);
elog(DEBUG1, "replaying changes from file \"%s\"", path);
- fd = BufFileOpenFileSet(MyLogicalRepWorker->stream_fileset, path, O_RDONLY,
- false);
+ stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
+ stream_xid = xid;
Why do we need stream_xid here? I think we can avoid having global
stream_fd if the comment #1 is feasible.
4.
+ * TRANS_LEADER_APPLY:
+ * The action means that we
/The/This. Please make a similar change for other actions.
5. Apart from the above, please find a few changes to the comments for
0001 and 0002 patches in the attached patches.
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] changes_amit_v54_0001.patch (3.3K, ../CAA4eK1Jwh7j86Egk1cye=x2R_yrTjzXGj7Fx12wVybBAEq91kA@mail.gmail.com/2-changes_amit_v54_0001.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index f9dda6daee..7c2b0e0b5e 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -64,12 +64,13 @@
* etc.) of the publisher and subscriber could be different, applying
* transactions in parallel mode on the subscriber side can cause some
* deadlocks that do not occur on the publisher side which is expected and can
- * happen even without parallel mode. In order to detect the deadlocks among
- * leader and parallel apply workers, we need to ensure that we wait using lmgr
- * locks, otherwise, such deadlocks won't be detected. The other approach was
- * to not allow parallelism when the schema of tables is different between the
- * publisher and subscriber but that would be too restrictive and would require
- * the publisher to send much more information than it is currently sending.
+ * happen even without parallel mode when there are concurrent operations on
+ * the subscriber. In order to detect the deadlocks among leader and parallel
+ * apply workers, we need to ensure that we wait using lmgr locks, otherwise,
+ * such deadlocks won't be detected. The other approach was to not allow
+ * parallelism when the schema of tables is different between the publisher
+ * and subscriber but that would be too restrictive and would require the
+ * publisher to send much more information than it is currently sending.
*
* Consider a case where the subscribed table does not have a unique key on the
* publisher and has a unique key on the subscriber.
@@ -1308,6 +1309,10 @@ pa_start_subtrans(TransactionId current_xid, TransactionId top_xid)
void
pa_clean_subtrans(void)
{
+ /*
+ * We don't need to free this explicitly as the allocated memory will be
+ * freed at the transaction end.
+ */
subxactlist = NIL;
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 997a1ef2d3..9510dc5683 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -484,6 +484,7 @@ should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
return MyLogicalRepWorker->relid == rel->localreloid;
else if (am_parallel_apply_worker())
{
+ /* We don't synchronize rel's that are in unknown state. */
if (rel->state != SUBREL_STATE_READY &&
rel->state != SUBREL_STATE_UNKNOWN)
ereport(ERROR,
@@ -1945,6 +1946,7 @@ apply_handle_stream_commit(StringInfo s)
pa_wait_for_xact_finish(winfo);
store_flush_position(commit_data.end_lsn, winfo->shared->last_commit_end);
+
(void) pa_free_worker(winfo, xid);
pgstat_report_stat(false);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 9b2ca9461b..c10b503e76 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -124,7 +124,10 @@ typedef struct ParallelApplyWorkerShared
*/
pg_atomic_uint32 pending_stream_count;
- /* XactLastCommitEnd from the parallel apply worker. */
+ /*
+ * XactLastCommitEnd from the parallel apply worker. This is required to
+ * update the lsn_mappings by leader worker.
+ */
XLogRecPtr last_commit_end;
} ParallelApplyWorkerShared;
[application/octet-stream] changes_amit_v54_0002.patch (3.7K, ../CAA4eK1Jwh7j86Egk1cye=x2R_yrTjzXGj7Fx12wVybBAEq91kA@mail.gmail.com/3-changes_amit_v54_0002.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 7eed82ff41..9e35b2ecbe 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -136,11 +136,11 @@
* concurrently), if the shm_mq buffer between LA and PA-2 is full, LA has to
* wait to send messages, and this wait doesn't appear in lmgr.
*
- * To resolve this issue, we use non-blocking write and wait with a timeout. If
+ * To avoid this wait, we use a non-blocking write and wait with a timeout. If
* the timeout is exceeded, the LA will serialize the message to a file and
* indicate PA-2 that it needs to read that file for the remaining messages.
* Then LA will start waiting for commit which will detect deadlock if any.
- * (See pa_send_data() and enum TransApplyAction)
+ * See pa_send_data() and enum TransApplyAction.
*
* 4) Lock types
*
@@ -691,10 +691,11 @@ pa_spooled_messages(void)
SpinLockRelease(&MyParallelShared->mutex);
/*
- * If the leader apply worker is still (busy) serializing partial changes
- * then the parallel apply worker acquires the stream lock now. Otherwise,
- * it would not have a chance to receive a STREAM_STOP (and acquire the
- * stream lock) until the leader had serialized all changes.
+ * If the leader apply worker is busy serializing the partial changes then
+ * acquire the stream lock now. Otherwise, the parallel apply worker won't
+ * get a chance to receive a STREAM_STOP (and acquire the stream lock)
+ * until the leader had serialized all changes which can lead to undetected
+ * deadlock.
*/
if (fileset_state == FS_BUSY)
{
@@ -1535,9 +1536,9 @@ pa_stream_abort(LogicalRepStreamAbortData *abort_data)
}
/*
- * Set the fileset state for the given parallel apply worker. The
- * stream_fileset of the leader apply worker will be written into the shared
- * memory if the fileset_state is FS_READY.
+ * Set the fileset state for the given parallel apply worker. The fileset
+ * will be set once the leader worker reached the ready state so that it can
+ * be used by parallel apply worker.
*/
void
pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b8587dbe5c..606e4c6fb4 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -256,15 +256,16 @@ typedef struct ApplyErrorCallbackArg
*
* TRANS_LEADER_PARTIAL_SERIALIZE:
* The action means that we are in the leader apply worker and have sent some
- * changes directly to the parallel apply worker, due to timeout while sending
- * data the remaining changes need to be serialized to a file. The parallel apply
- * worker will apply these serialized changes when the final commit arrives.
+ * changes directly to the parallel apply worker and the remaining
+ * changes are serialized to a file, due to timeout while sending data. The
+ * parallel apply worker will apply these serialized changes when the final
+ * commit arrives.
*
- * LEADER_SERIALIZE could not be used for this case because, in addition to
+ * We can't use TRANS_LEADER_SERIALIZE for this case because, in addition to
* serializing changes, the leader worker also needs to serialize the STREAM_XXX
* message to a file, and wait for the parallel apply worker to finish the
* transaction when processing the transaction finish command. So this new
- * action was introduced to make the logic clearer.
+ * action was introduced to keep the code and logic clear.
*
* TRANS_LEADER_SEND_TO_PARALLEL:
* The action means that we are in the leader apply worker and need to send the
view thread (625+ messages) latest in thread
reply
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: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <CAA4eK1Jwh7j86Egk1cye=x2R_yrTjzXGj7Fx12wVybBAEq91kA@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox