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: 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-05-31T08:52:41Z
Lists: pgsql-hackers 
On Mon, May 30, 2022 at 5:08 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
>
> On Mon, May 30, 2022 at 2:22 PM wangw.fnst@fujitsu.com
> <wangw.fnst@fujitsu.com> wrote:
> >
> > Attach the new patches(only changed 0001 and 0002)
> >
>

This patch allows the same replication origin to be used by the main
apply worker and the bgworker that uses it to apply streaming
transactions. See the changes [1] in the patch. I am not completely
sure whether that is a good idea even though I could not spot or think
of problems that can't be fixed in your patch. I see that currently
both the main apply worker and bgworker will assign MyProcPid to the
assigned origin slot, this can create the problem because
ReplicationOriginExitCleanup() can clean it up even though the main
apply worker or another bgworker is still using that origin slot. Now,
one way to fix is that we assign only the main apply worker's
MyProcPid to session_replication_state->acquired_by. I have tried to
think about the concurrency issues as multiple workers could now point
to the same replication origin state. I think it is safe because the
patch maintains the commit order by allowing only one process to
commit at a time, so no two workers will be operating on the same
origin at the same time. Even, though there is no case where the patch
will try to advance the session's origin concurrently, it appears safe
to do so as we change/advance the session_origin LSNs under
replicate_state LWLock.

Another idea could be that we allow multiple replication origins (one
for each bgworker and one for the main apply worker) for the apply
workers corresponding to a subscription. Then on restart, we can find
the highest LSN among all the origins for a subscription. This should
work primarily because we will maintain the commit order. Now, for
this to work we need to somehow map all the origins for a subscription
and one possibility is that we have a subscription id in each of the
origin names. Currently we use ("pg_%u", MySubscription->oid) as
origin_name. We can probably append some unique identifier number for
each worker to allow each origin to have a subscription id. We need to
drop all origins for a particular subscription on DROP SUBSCRIPTION. I
think having multiple origins for the same subscription will have some
additional work when we try to filter changes based on origin.

The advantage of the first idea is that it won't increase the need to
have more origins per subscription but it is quite possible that I am
missing something and there are problems due to which we can't use
that approach.

Thoughts?

[1]:
-replorigin_session_setup(RepOriginId node)
+replorigin_session_setup(RepOriginId node, bool acquire)
 {
  static bool registered_cleanup;
  int i;
@@ -1110,7 +1110,7 @@ replorigin_session_setup(RepOriginId node)
  if (curstate->roident != node)
  continue;

- else if (curstate->acquired_by != 0)
+ else if (curstate->acquired_by != 0 && acquire)
  {
...
...

+ /*
+ * The apply bgworker don't need to monopolize this replication origin
+ * which was already acquired by its leader process.
+ */
+ replorigin_session_setup(originid, false);
+ replorigin_session_origin = originid;

-- 
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.