RE: Perform streaming logical transactions by background workers and parallel apply
Yu Shi (Fujitsu) <shiy.fnst@fujitsu.com>
On Thu, Aug 4, 2022 2:36 PM Wang, Wei/王 威 <wangw.fnst@fujitsu.com> wrote:
>
> I also did some other improvements based on the suggestions posted in this
> thread. Attach the new patches.
>
Thanks for updating the patch. Here are some comments on v20-0001 patch.
1.
+typedef struct ApplyBgworkerShared
+{
+ slock_t mutex;
+
+ /* Status of apply background worker. */
+ ApplyBgworkerStatus status;
+
+ /* proto version of publisher. */
+ uint32 proto_version;
+
+ TransactionId stream_xid;
+
+ /* id of apply background worker */
+ uint32 worker_id;
+} ApplyBgworkerShared;
Would it be better to modify the comment of "proto_version" to "Logical protocol
version"?
2. commnent of handle_streamed_transaction()
+ * Exception: When the main apply worker is applying streaming transactions in
+ * parallel mode (e.g. when addressing LOGICAL_REP_MSG_RELATION or
+ * LOGICAL_REP_MSG_TYPE changes), then return false.
This comment doesn't look very clear, could we change it to:
Exception: In SUBSTREAM_PARALLEL mode, if the message type is
LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE, return false even if this is
the main apply worker.
3.
+/*
+ * There are three fields in message: start_lsn, end_lsn and send_time. Because
+ * we have updated these statistics in apply worker, we could ignore these
+ * fields in apply background worker. (see function LogicalRepApplyLoop)
+ */
+#define SIZE_STATS_MESSAGE (3 * sizeof(uint64))
updated these statistics in apply worker
->
updated these statistics in main apply worker
4.
+static void
+LogicalApplyBgwLoop(shm_mq_handle *mqh, volatile ApplyBgworkerShared *shared)
+{
+ shm_mq_result shmq_res;
+ PGPROC *registrant;
+ ErrorContextCallback errcallback;
I think we can define "shmq_res" in the for loop.
5.
+ /*
+ * We use first byte of message for additional communication between
+ * main Logical replication worker and apply background workers, so if
+ * it differs from 'w', then process it first.
+ */
between main Logical replication worker and apply background workers
->
between main apply worker and apply background workers
Regards,
Shi yu
Commits
GET /api/v1/messages/:b64id/commits
the thread's linked commits as JSON, with link sources.
API reference →
-
Fix invalid memory access during the shutdown of the parallel apply worker.
- 3d144c6c8602 16.0 landed
-
Fix assertion failure in apply worker.
- de63f8dadee4 16.0 landed
-
Use elog to report unexpected action in handle_streamed_transaction().
- 781ac42d43ab 16.0 landed
-
Use appropriate wait event when sending data in the apply worker.
- d9d7fe68d35e 16.0 landed
-
Allow the logical_replication_mode to be used on the subscriber.
- 9f2213a7c575 16.0 landed
-
Rename GUC logical_decoding_mode to logical_replication_mode.
- 1e8b61735cfb 16.0 landed
-
Display the leader apply worker's PID for parallel apply workers.
- d540a02a724b 16.0 landed
-
Improve the code to decide and process the apply action.
- c981d9145dea 16.0 landed
-
Document the newly added wait events added by commit 216a784829.
- cd06ccd78fcf 16.0 landed
-
Perform apply of large transactions by parallel workers.
- 216a784829c2 16.0 landed
-
Wake up a subscription's replication worker processes after DDL.
- c6e1f62e2cee 16.0 cited
-
Add copyright notices to meson files
- 8284cf5f746f 16.0 cited
-
Better document logical replication parameters
- a8500750ca0a 16.0 cited
-
Add a common function to generate the origin name.
- 776e1c8a5d14 16.0 cited
-
Harmonize parameter names in storage and AM code.
- bfcf1b34805f 16.0 cited
-
Avoid using list_length() to test for empty list.
- efd0c16becbf 16.0 cited
-
Improve two comments related to a boolean DefElem's value
- 8445f5a21d40 16.0 cited
-
Fix partition table's REPLICA IDENTITY checking on the subscriber.
- 26b3455afaa9 15.0 cited
-
Fix data inconsistency between publisher and subscriber.
- b7658c24c7c1 15.0 cited
-
Fix cache look-up failures while applying changes in logical replication.
- 5a97b1325453 15.0 cited