Horton
Menu

Re: Perform streaming logical transactions by background workers and parallel apply

Amit Kapila <amit.kapila16@gmail.com>

View thread
From: Amit Kapila <amit.kapila16@gmail.com>
To: "wangw.fnst@fujitsu.com" <wangw.fnst@fujitsu.com>
Cc: Masahiko Sawada <sawada.mshk@gmail.com>, Peter Smith <smithpb2250@gmail.com>, "shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>, "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>, PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Date: 2022-06-17T08:57:45Z
Lists: pgsql-hackers 
On Fri, Jun 17, 2022 at 12:47 PM wangw.fnst@fujitsu.com
<wangw.fnst@fujitsu.com> wrote:
>
> Attach the new patches.
> Only changed patches 0001, 0004.
>

Few more comments on the previous version of patch:
===========================================
1.
+/*
+ * Count the number of registered (not necessarily running) apply background
+ * worker for a subscription.
+ */

/worker/workers

2.
+static void
+apply_bgworker_setup_dsm(ApplyBgworkerState *wstate)
+{
...
...
+ int64 queue_size = 160000000; /* 16 MB for now */

I think it would be better to use define for this rather than a
hard-coded value.

3.
+/*
+ * Status for apply background worker.
+ */
+typedef enum ApplyBgworkerStatus
+{
+ APPLY_BGWORKER_ATTACHED = 0,
+ APPLY_BGWORKER_READY,
+ APPLY_BGWORKER_BUSY,
+ APPLY_BGWORKER_FINISHED,
+ APPLY_BGWORKER_EXIT
+} ApplyBgworkerStatus;

It would be better if you can add comments to explain each of these states.

4.
+ /* Set up one message queue per worker, plus one. */
+ mq = shm_mq_create(shm_toc_allocate(toc, (Size) queue_size),
+    (Size) queue_size);
+ shm_toc_insert(toc, APPLY_BGWORKER_KEY_MQ, mq);
+ shm_mq_set_sender(mq, MyProc);


I don't understand the meaning of 'plus one' in the above comment as
the patch seems to be setting up just one queue here?

5.
+
+ /* Attach the queues. */
+ wstate->mq_handle = shm_mq_attach(mq, seg, NULL);

Similar to above. If there is only one queue then the comment should
say queue instead of queues.

6.
  snprintf(bgw.bgw_name, BGW_MAXLEN,
  "logical replication worker for subscription %u", subid);
+ else
+ snprintf(bgw.bgw_name, BGW_MAXLEN,
+ "logical replication background apply worker for subscription %u ", subid);

No need for extra space after %u in the above code.

7.
+ launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+ MySubscription->oid,
+ MySubscription->name,
+ MyLogicalRepWorker->userid,
+ InvalidOid,
+ dsm_segment_handle(wstate->dsm_seg));
+
+ if (launched)
+ {
+ /* Wait for worker to attach. */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_ATTACHED);

In logicalrep_worker_launch(), we already seem to be waiting for
workers to attach via WaitForReplicationWorkerAttach(), so it is not
clear to me why we need to wait again? If there is a genuine reason
then it is better to add some comments to explain it. I think in some
way, we need to know if the worker is successfully attached and we may
not get that via WaitForReplicationWorkerAttach, so there needs to be
some way to know that but this doesn't sound like a very good idea. If
that understanding is correct then can we think of a better way?

8. I think we can simplify apply_bgworker_find_or_start by having
separate APIs for find and start. Most of the places need to use find
API except for the first stream. If we do that then I think you don't
need to make a hash entry unless we established ApplyBgworkerState
which currently looks odd as you need to remove the entry if we fail
to allocate the state.

9.
+ /*
+ * TO IMPROVE: Do we need to display the apply background worker's
+ * information in pg_stat_replication ?
+ */
+ UpdateWorkerStats(last_received, send_time, false);

In this do you mean to say pg_stat_subscription? If so, then to decide
whether we need to update stats here we should see what additional
information we can update here which is not possible via the main
apply worker?

10.
ApplyBgworkerMain
{
...
+ /* Load the subscription into persistent memory context. */
+ ApplyContext = AllocSetContextCreate(TopMemoryContext,
...

This comment seems to be copied from ApplyWorkerMain but doesn't apply here.

-- 
With Regards,
Amit Kapila.

Commits

Same data as JSON: GET /api/v1/messages/:b64id/commits the thread's linked commits as JSON, with link sources. API reference →

  1. Fix invalid memory access during the shutdown of the parallel apply worker.

  2. Fix assertion failure in apply worker.

  3. Use elog to report unexpected action in handle_streamed_transaction().

  4. Use appropriate wait event when sending data in the apply worker.

  5. Allow the logical_replication_mode to be used on the subscriber.

  6. Rename GUC logical_decoding_mode to logical_replication_mode.

  7. Display the leader apply worker's PID for parallel apply workers.

  8. Improve the code to decide and process the apply action.

  9. Document the newly added wait events added by commit 216a784829.

  10. Perform apply of large transactions by parallel workers.

  11. Wake up a subscription's replication worker processes after DDL.

  12. Add copyright notices to meson files

  13. Better document logical replication parameters

  14. Add a common function to generate the origin name.

  15. Harmonize parameter names in storage and AM code.

  16. Avoid using list_length() to test for empty list.

  17. Improve two comments related to a boolean DefElem's value

  18. Fix partition table's REPLICA IDENTITY checking on the subscriber.

  19. Fix data inconsistency between publisher and subscriber.

  20. Fix cache look-up failures while applying changes in logical replication.