public inbox for [email protected]
help / color / mirror / Atom feedFrom: Amit Kapila <[email protected]>
To: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Masahiko Sawada <[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, 15 Sep 2022 17:09:45 +0530
Message-ID: <CAA4eK1LMVdS6uM7Tw7ANL0BetAd76TKkmAXNNQa0haTe2tax6g@mail.gmail.com> (raw)
In-Reply-To: <OS3PR01MB6275F145878B4A44586C46CE9E499@OS3PR01MB6275.jpnprd01.prod.outlook.com>
References: <OS0PR01MB5716730C06159452335D870D947F9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1Lu-6oXMk7ZaGYLwm3CRLBuzueGbasyHnNpJxu6Mq3mmg@mail.gmail.com>
<OS3PR01MB6275F145878B4A44586C46CE9E499@OS3PR01MB6275.jpnprd01.prod.outlook.com>
On Thu, Sep 15, 2022 at 10:45 AM [email protected]
<[email protected]> wrote:
>
> Attach the new patch set.
>
Review of v29-0001*
==================
1.
+parallel_apply_find_worker(TransactionId xid)
{
...
+ entry = hash_search(ParallelApplyWorkersHash, &xid, HASH_FIND, &found);
+ if (found)
+ {
+ /* If any workers (or the postmaster) have died, we have failed. */
+ if (entry->winfo->error_mq_handle == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("lost connection to parallel apply worker")));
...
}
I think the above comment is incorrect because if the postmaster would
have died then you wouldn't have found the entry in the hash table.
How about something like: "We can't proceed if the parallel streaming
worker has already exited."
2.
+/*
+ * Find the previously assigned worker for the given transaction, if any.
+ */
+ParallelApplyWorkerInfo *
+parallel_apply_find_worker(TransactionId xid)
No need to use word 'previously' in the above sentence.
3.
+ * We need one key to register the location of the header, and we need
+ * another key to track the location of the message queue.
+ */
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(ParallelApplyWorkerShared));
+ shm_toc_estimate_chunk(&e, queue_size);
+ shm_toc_estimate_chunk(&e, error_queue_size);
+
+ shm_toc_estimate_keys(&e, 3);
Overall, three keys are used but the comment indicates two. You forgot
to mention about error_queue.
4.
+ if (launched)
+ ParallelApplyWorkersList = lappend(ParallelApplyWorkersList, winfo);
+ else
+ {
+ shm_mq_detach(winfo->mq_handle);
+ shm_mq_detach(winfo->error_mq_handle);
+ dsm_detach(winfo->dsm_seg);
+ pfree(winfo);
+
+ winfo = NULL;
+ }
A. The code used in the else part to free worker info is the same as
what is used in parallel_apply_free_worker. Can we move this to a
separate function say parallel_apply_free_worker_info()?
B. I think it will be better if you use {} for if branch to make it
look consistent with else branch.
5.
+ * case define a named savepoint, so that we are able to commit/rollback it
+ * separately later.
+ */
+void
+parallel_apply_subxact_info_add(TransactionId current_xid)
I don't see the need of commit in the above message. So, we can
slightly modify it to: "... so that we are able to rollback to it
separately later."
6.
+ for (i = list_length(subxactlist) - 1; i >= 0; i--)
+ {
+ xid = list_nth_xid(subxactlist, i);
...
...
+/*
+ * Return the TransactionId value contained in the n'th element of the
+ * specified list.
+ */
+static inline TransactionId
+list_nth_xid(const List *list, int n)
+{
+ Assert(IsA(list, XidList));
+ return lfirst_xid(list_nth_cell(list, n));
+}
I am not really sure that we need a new list function to use for this
place. Can't we directly use lfirst_xid(list_nth_cell) instead?
7.
+void
+parallel_apply_replorigin_setup(void)
+{
+ RepOriginId originid;
+ char originname[NAMEDATALEN];
+ bool started_tx = false;
+
+ /* This function might be called inside or outside of transaction. */
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
Is there a place in the patch where this function will be called
without having an active transaction state? If so, then this coding is
fine but if not, then I suggest keeping an assert for transaction
state here. The same thing applies to
parallel_apply_replorigin_reset() as well.
8.
+ *
+ * If write_abort_lsn is true, send the abort_lsn and abort_time fields,
+ * otherwise don't.
*/
void
logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
- TransactionId subxid)
+ TransactionId subxid, XLogRecPtr abort_lsn,
+ TimestampTz abort_time, bool abort_info)
In the comment, the name of the variable needs to be updated.
9.
+TransactionId stream_xid = InvalidTransactionId;
-static TransactionId stream_xid = InvalidTransactionId;
...
...
+void
+parallel_apply_subxact_info_add(TransactionId current_xid)
+{
+ if (current_xid != stream_xid &&
+ !list_member_xid(subxactlist, current_xid))
It seems you have changed the scope of stream_xid to use it in
parallel_apply_subxact_info_add(). Won't it be better to pass it as a
parameter (say top_xid)?
10.
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -20,6 +20,7 @@
#include <sys/time.h>
#include "access/xlog.h"
+#include "catalog/pg_subscription.h"
#include "catalog/pg_type.h"
#include "common/connect.h"
#include "funcapi.h"
@@ -443,9 +444,14 @@ libpqrcv_startstreaming(WalReceiverConn *conn,
appendStringInfo(&cmd, "proto_version '%u'",
options->proto.logical.proto_version);
- if (options->proto.logical.streaming &&
- PQserverVersion(conn->streamConn) >= 140000)
- appendStringInfoString(&cmd, ", streaming 'on'");
+ if (options->proto.logical.streaming != SUBSTREAM_OFF)
+ {
+ if (PQserverVersion(conn->streamConn) >= 160000 &&
+ options->proto.logical.streaming == SUBSTREAM_PARALLEL)
+ appendStringInfoString(&cmd, ", streaming 'parallel'");
+ else if (PQserverVersion(conn->streamConn) >= 140000)
+ appendStringInfoString(&cmd, ", streaming 'on'");
+ }
It doesn't seem like a good idea to expose subscription options here.
Can we think of having char *streaming_option instead of the current
streaming parameter which is filled by the caller and used here
directly?
11. The error message used in pgoutput_startup() seems to be better
than the current messages used in that function but it is better to be
consistent with other messages. There is a discussion in the email
thread [1] on improving those messages, so kindly suggest there.
12. In addition to the above, I have changed/added a few comments in
the attached patch.
[1] - https://www.postgresql.org/message-id/20220914.111507.13049297635620898.horikyota.ntt%40gmail.com
--
With Regards,
Amit Kapila.
Attachments:
[application/octet-stream] changed_comments_amit_v29.patch (2.6K, ../CAA4eK1LMVdS6uM7Tw7ANL0BetAd76TKkmAXNNQa0haTe2tax6g@mail.gmail.com/2-changed_comments_amit_v29.patch)
download | inline diff:
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 8344e4509b..c154eed34c 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -664,15 +664,13 @@ logicalrep_worker_attach(int slot)
}
/*
- * Detach the worker (cleans up the worker info).
+ * Stop the parallel apply workers if any, and detach the leader apply
+ * worker (cleans up the worker info).
*/
static void
logicalrep_worker_detach(void)
{
- /*
- * If this is the leader apply worker then stop all the parallel apply
- * workers previously started from here.
- */
+ /* Stop the parallel apply workers. */
if (!am_parallel_apply_worker() && !am_tablesync_worker())
{
List *workers;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 2bff1dbc74..e8aa661971 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -58,20 +58,18 @@
* parallel apply workers (max_parallel_apply_workers_per_subscription), but this
* may waste some memory if no process is actually started.
*
- * The dynamic shared memory segment will contain (1) a shm_mq that can be
+ * The dynamic shared memory segment will contain (a) a shm_mq that can be
* used to send changes in the transaction from leader apply worker to parallel
- * apply worker (2) another shm_mq that can be used to send errors (and other
+ * apply worker (b) another shm_mq that can be used to send errors (and other
* messages reported via elog/ereport) from the parallel apply worker to leader
- * apply worker (3) necessary information to be shared among parallel apply
- * workers and leader apply worker(i.e. in_parallel_apply_xact flag and the
+ * apply worker (c) necessary information to be shared among parallel apply
+ * workers and leader apply worker (i.e. in_parallel_apply_xact flag and the
* corresponding LogicalRepWorker slot information).
*
- * 2) Write to temporary files and apply when the final commit arrives
+ * In case, no worker is available to handle the streamed transaction, we
+ * follow approach 2.
*
- * If streaming = true, or if streaming = parallel but there are not parallel
- * apply workers available to handle the streamed transaction, the data is
- * written to temporary files and then applied at once when the final commit
- * arrives.
+ * 2) Write to temporary files and apply when the final commit arrives
*
* Unlike the regular (non-streamed) case, handling streamed transactions has
* to handle aborts of both the toplevel transaction and subtransactions. This
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: <CAA4eK1LMVdS6uM7Tw7ANL0BetAd76TKkmAXNNQa0haTe2tax6g@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