RE: Perform streaming logical transactions by background workers and parallel apply
Zhijie Hou (Fujitsu) <houzj.fnst@fujitsu.com>
Attachments
- v62-0008-Add-a-main_worker_pid-to-pg_stat_subscription.patch (application/octet-stream) patch v62-0008
- v62-0002-Serialize-partial-changes-to-a-file-when-the-att.patch (application/octet-stream) patch v62-0002
- v62-0001-Perform-streaming-logical-transactions-by-parall.patch (application/octet-stream) patch v62-0001
- v62-0006-Stop-extra-worker-if-GUC-was-changed.patch (application/octet-stream) patch v62-0006
- v62-0003-Test-streaming-parallel-option-in-tap-test.patch (application/octet-stream) patch v62-0003
- v62-0004-Allow-streaming-every-change-without-waiting-til.patch (application/octet-stream) patch v62-0004
- v62-0005-Add-GUC-stream_serialize_threshold-and-test-seri.patch (application/octet-stream) patch v62-0005
- v62-0007-Retry-to-apply-streaming-xact-only-in-apply-work.patch (application/octet-stream) patch v62-0007
On Saturday, December 17, 2022 8:16 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
>
> On Fri, Dec 16, 2022 at 4:34 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
> >
> > On Fri, Dec 16, 2022 at 2:47 PM houzj.fnst@fujitsu.com
> > <houzj.fnst@fujitsu.com> wrote:
> > >
> > > > ---
> > > > + active_workers = list_copy(ParallelApplyWorkerPool);
> > > > +
> > > > + foreach(lc, active_workers)
> > > > + {
> > > > + int slot_no;
> > > > + uint16 generation;
> > > > + ParallelApplyWorkerInfo *winfo =
> > > > (ParallelApplyWorkerInfo *) lfirst(lc);
> > > > +
> > > > + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
> > > > + napplyworkers =
> > > > logicalrep_pa_worker_count(MyLogicalRepWorker->subid);
> > > > + LWLockRelease(LogicalRepWorkerLock);
> > > > +
> > > > + if (napplyworkers <=
> > > > max_parallel_apply_workers_per_subscription / 2)
> > > > + return;
> > > > +
> > > >
> > > > Calling logicalrep_pa_worker_count() with lwlock for each worker
> > > > seems not efficient to me. I think we can get the number of
> > > > workers once at the top of this function and return if it's
> > > > already lower than the maximum pool size. Otherwise, we attempt to stop
> extra workers.
> > >
> > > How about we directly check the length of worker pool list here
> > > which seems simpler and don't need to lock ?
> > >
> >
> > I don't see any problem with that. Also, if such a check is safe then
> > can't we use the same in pa_free_worker() as well? BTW, shouldn't
> > pa_stop_idle_workers() try to free/stop workers unless the active
> > number reaches below max_parallel_apply_workers_per_subscription?
> >
>
> BTW, can we move pa_stop_idle_workers() functionality to a later patch (say into
> v61-0006*)? That way we can focus on it separately once the main patch is
> committed.
Agreed. I have addressed all the comments and did some cosmetic changes.
Attach the new version patch set.
Best regards,
Hou zj
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