Horton
Menu

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

Zhijie Hou (Fujitsu) <houzj.fnst@fujitsu.com>

View thread
From: "houzj.fnst@fujitsu.com" <houzj.fnst@fujitsu.com>
To: Amit Kapila <amit.kapila16@gmail.com>
Cc: Masahiko Sawada <sawada.mshk@gmail.com>, "wangw.fnst@fujitsu.com" <wangw.fnst@fujitsu.com>, Peter Smith <smithpb2250@gmail.com>, Dilip Kumar <dilipbalaut@gmail.com>, "shiy.fnst@fujitsu.com" <shiy.fnst@fujitsu.com>, PostgreSQL Hackers <pgsql-hackers@lists.postgresql.org>
Date: 2022-11-29T04:48:28Z
Lists: pgsql-hackers

Attachments

On Mon, November 28, 2022 20:26 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
> On Sun, Nov 27, 2022 at 9:43 AM houzj.fnst@fujitsu.com 
> <houzj.fnst@fujitsu.com> wrote:
> >
> > Attach the new version patch which addressed all comments so far.
> >
> 
> Few comments on v52-0001*
> ========================
> 1.
> pa_free_worker()
> {
> ...
> + /* Free the worker information if the worker exited cleanly. */ if 
> + (!winfo->error_mq_handle) { pa_free_worker_info(winfo);
> +
> + if (winfo->in_use &&
> + !hash_search(ParallelApplyWorkersHash, &xid, HASH_REMOVE, NULL)) 
> + elog(ERROR, "hash table corrupted");
> 
> pa_free_worker_info() pfrees the winfo, so how is it legal to
> winfo->in_use in the above check?
> 
> Also, why is this check (!winfo->error_mq_handle) required in the 
> first place in the patch? The worker exits cleanly only when the 
> leader apply worker sends a SIGINT signal and in that case, we already 
> detach from the error queue and clean up other worker information.

It was intended for the case when a user send a signal, but it seems not standard way to do that.
So, I removed this check (!winfo->error_mq_handle).

> 2.
> +HandleParallelApplyMessages(void)
> +{
> ...
> ...
> + foreach(lc, ParallelApplyWorkersList) { shm_mq_result res; Size 
> + nbytes;
> + void    *data;
> + ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) 
> + lfirst(lc);
> +
> + if (!winfo->error_mq_handle)
> + continue;
> 
> Similar to the previous comment, it is not clear whether we need this 
> check. If required, can we add a comment to indicate the case where it 
> happens to be true?
> Note, there is a similar check for winfo->error_mq_handle in 
> pa_wait_for_xact_state(). Please add some comments if that is 
> required.

Removed this check in these two functions.

> 3. Why is there apply_worker_clean_exit() at the end of 
> ParallelApplyWorkerMain()? Normally either the leader worker stops 
> parallel apply, or parallel apply gets stopped because of a parameter 
> change, or exits because of error, and in none of those cases it can 
> hit this code path unless I am missing something.
> 
> Additionally, I think in LogicalParallelApplyLoop, we will never 
> receive zero-length messages so that is also wrong and should be 
> converted to elog(ERROR,..).

Agreed and changed. 

> 4. I think in logicalrep_worker_detach(), we should detach from the 
> shm error queue so that the parallel apply worker won't try to send a 
> termination message back to the leader worker.

Agreed and changed.

> 5.
> pa_send_data()
> {
> ...
> + if (startTime == 0)
> + startTime = GetCurrentTimestamp();
> ...
> 
> What is the use of getting the current timestamp before waitlatch 
> logic, if it is not used before that? It seems that is for the time 
> logic to look correct. We can probably reduce the 10s interval to 9s 
> for that.

Changed.

> In this function, we need to add some comments to indicate why the 
> current logic is used, and also probably we can refer to the comments 
> atop this file.

Added some comments.

> 6. I think it will be better if we keep stream_apply_worker local to 
> applyparallelworker.c by exposing functions to cache/resetting the 
> required info.

Agree. Added a new function to set the stream_apply_worker.

> 7. Apart from the above, I have made a few changes in the comments and 
> some miscellaneous cosmetic changes in the attached. Kindly include 
> these in the next version unless you see a problem with any change.

Thanks, I have checked and merge them.

Attach the new version patch which addressed all comments.

Best regards,
Hou zj

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.