Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * applyparallelworker.c
3 : : * Support routines for applying xact by parallel apply worker
4 : : *
5 : : * Copyright (c) 2023-2026, PostgreSQL Global Development Group
6 : : *
7 : : * IDENTIFICATION
8 : : * src/backend/replication/logical/applyparallelworker.c
9 : : *
10 : : * This file contains the code to launch, set up, and teardown a parallel apply
11 : : * worker which receives the changes from the leader worker and invokes routines
12 : : * to apply those on the subscriber database. Additionally, this file contains
13 : : * routines that are intended to support setting up, using, and tearing down a
14 : : * ParallelApplyWorkerInfo which is required so the leader worker and parallel
15 : : * apply workers can communicate with each other.
16 : : *
17 : : * The parallel apply workers are assigned (if available) as soon as xact's
18 : : * first stream is received for subscriptions that have set their 'streaming'
19 : : * option as parallel. The leader apply worker will send changes to this new
20 : : * worker via shared memory. We keep this worker assigned till the transaction
21 : : * commit is received and also wait for the worker to finish at commit. This
22 : : * preserves commit ordering and avoid file I/O in most cases, although we
23 : : * still need to spill to a file if there is no worker available. See comments
24 : : * atop logical/worker to know more about streamed xacts whose changes are
25 : : * spilled to disk. It is important to maintain commit order to avoid failures
26 : : * due to: (a) transaction dependencies - say if we insert a row in the first
27 : : * transaction and update it in the second transaction on publisher then
28 : : * allowing the subscriber to apply both in parallel can lead to failure in the
29 : : * update; (b) deadlocks - allowing transactions that update the same set of
30 : : * rows/tables in the opposite order to be applied in parallel can lead to
31 : : * deadlocks.
32 : : *
33 : : * A worker pool is used to avoid restarting workers for each streaming
34 : : * transaction. We maintain each worker's information (ParallelApplyWorkerInfo)
35 : : * in the ParallelApplyWorkerPool. After successfully launching a new worker,
36 : : * its information is added to the ParallelApplyWorkerPool. Once the worker
37 : : * finishes applying the transaction, it is marked as available for re-use.
38 : : * Now, before starting a new worker to apply the streaming transaction, we
39 : : * check the list for any available worker. Note that we retain a maximum of
40 : : * half the max_parallel_apply_workers_per_subscription workers in the pool and
41 : : * after that, we simply exit the worker after applying the transaction.
42 : : *
43 : : * XXX This worker pool threshold is arbitrary and we can provide a GUC
44 : : * variable for this in the future if required.
45 : : *
46 : : * The leader apply worker will create a separate dynamic shared memory segment
47 : : * when each parallel apply worker starts. The reason for this design is that
48 : : * we cannot predict how many workers will be needed. It may be possible to
49 : : * allocate enough shared memory in one segment based on the maximum number of
50 : : * parallel apply workers (max_parallel_apply_workers_per_subscription), but
51 : : * this would waste memory if no process is actually started.
52 : : *
53 : : * The dynamic shared memory segment contains: (a) a shm_mq that is used to
54 : : * send changes in the transaction from leader apply worker to parallel apply
55 : : * worker; (b) another shm_mq that is used to send errors (and other messages
56 : : * reported via elog/ereport) from the parallel apply worker to leader apply
57 : : * worker; (c) necessary information to be shared among parallel apply workers
58 : : * and the leader apply worker (i.e. members of ParallelApplyWorkerShared).
59 : : *
60 : : * Locking Considerations
61 : : * ----------------------
62 : : * We have a risk of deadlock due to concurrently applying the transactions in
63 : : * parallel mode that were independent on the publisher side but became
64 : : * dependent on the subscriber side due to the different database structures
65 : : * (like schema of subscription tables, constraints, etc.) on each side. This
66 : : * can happen even without parallel mode when there are concurrent operations
67 : : * on the subscriber. In order to detect the deadlocks among leader (LA) and
68 : : * parallel apply (PA) workers, we used lmgr locks when the PA waits for the
69 : : * next stream (set of changes) and LA waits for PA to finish the transaction.
70 : : * An alternative approach could be to not allow parallelism when the schema of
71 : : * tables is different between the publisher and subscriber but that would be
72 : : * too restrictive and would require the publisher to send much more
73 : : * information than it is currently sending.
74 : : *
75 : : * Consider a case where the subscribed table does not have a unique key on the
76 : : * publisher and has a unique key on the subscriber. The deadlock can happen in
77 : : * the following ways:
78 : : *
79 : : * 1) Deadlock between the leader apply worker and a parallel apply worker
80 : : *
81 : : * Consider that the parallel apply worker (PA) is executing TX-1 and the
82 : : * leader apply worker (LA) is executing TX-2 concurrently on the subscriber.
83 : : * Now, LA is waiting for PA because of the unique key constraint of the
84 : : * subscribed table while PA is waiting for LA to send the next stream of
85 : : * changes or transaction finish command message.
86 : : *
87 : : * In order for lmgr to detect this, we have LA acquire a session lock on the
88 : : * remote transaction (by pa_lock_stream()) and have PA wait on the lock before
89 : : * trying to receive the next stream of changes. Specifically, LA will acquire
90 : : * the lock in AccessExclusive mode before sending the STREAM_STOP and will
91 : : * release it if already acquired after sending the STREAM_START, STREAM_ABORT
92 : : * (for toplevel transaction), STREAM_PREPARE, and STREAM_COMMIT. The PA will
93 : : * acquire the lock in AccessShare mode after processing STREAM_STOP and
94 : : * STREAM_ABORT (for subtransaction) and then release the lock immediately
95 : : * after acquiring it.
96 : : *
97 : : * The lock graph for the above example will look as follows:
98 : : * LA (waiting to acquire the lock on the unique index) -> PA (waiting to
99 : : * acquire the stream lock) -> LA
100 : : *
101 : : * This way, when PA is waiting for LA for the next stream of changes, we can
102 : : * have a wait-edge from PA to LA in lmgr, which will make us detect the
103 : : * deadlock between LA and PA.
104 : : *
105 : : * 2) Deadlock between the leader apply worker and parallel apply workers
106 : : *
107 : : * This scenario is similar to the first case but TX-1 and TX-2 are executed by
108 : : * two parallel apply workers (PA-1 and PA-2 respectively). In this scenario,
109 : : * PA-2 is waiting for PA-1 to complete its transaction while PA-1 is waiting
110 : : * for subsequent input from LA. Also, LA is waiting for PA-2 to complete its
111 : : * transaction in order to preserve the commit order. There is a deadlock among
112 : : * the three processes.
113 : : *
114 : : * In order for lmgr to detect this, we have PA acquire a session lock (this is
115 : : * a different lock than referred in the previous case, see
116 : : * pa_lock_transaction()) on the transaction being applied and have LA wait on
117 : : * the lock before proceeding in the transaction finish commands. Specifically,
118 : : * PA will acquire this lock in AccessExclusive mode before executing the first
119 : : * message of the transaction and release it at the xact end. LA will acquire
120 : : * this lock in AccessShare mode at transaction finish commands and release it
121 : : * immediately.
122 : : *
123 : : * The lock graph for the above example will look as follows:
124 : : * LA (waiting to acquire the transaction lock) -> PA-2 (waiting to acquire the
125 : : * lock due to unique index constraint) -> PA-1 (waiting to acquire the stream
126 : : * lock) -> LA
127 : : *
128 : : * This way when LA is waiting to finish the transaction end command to preserve
129 : : * the commit order, we will be able to detect deadlock, if any.
130 : : *
131 : : * One might think we can use XactLockTableWait(), but XactLockTableWait()
132 : : * considers PREPARED TRANSACTION as still in progress which means the lock
133 : : * won't be released even after the parallel apply worker has prepared the
134 : : * transaction.
135 : : *
136 : : * 3) Deadlock when the shm_mq buffer is full
137 : : *
138 : : * In the previous scenario (ie. PA-1 and PA-2 are executing transactions
139 : : * concurrently), if the shm_mq buffer between LA and PA-2 is full, LA has to
140 : : * wait to send messages, and this wait doesn't appear in lmgr.
141 : : *
142 : : * To avoid this wait, we use a non-blocking write and wait with a timeout. If
143 : : * the timeout is exceeded, the LA will serialize all the pending messages to
144 : : * a file and indicate PA-2 that it needs to read that file for the remaining
145 : : * messages. Then LA will start waiting for commit as in the previous case
146 : : * which will detect deadlock if any. See pa_send_data() and
147 : : * enum TransApplyAction.
148 : : *
149 : : * Lock types
150 : : * ----------
151 : : * Both the stream lock and the transaction lock mentioned above are
152 : : * session-level locks because both locks could be acquired outside the
153 : : * transaction, and the stream lock in the leader needs to persist across
154 : : * transaction boundaries i.e. until the end of the streaming transaction.
155 : : *-------------------------------------------------------------------------
156 : : */
157 : :
158 : : #include "postgres.h"
159 : :
160 : : #include "libpq/pqformat.h"
161 : : #include "libpq/pqmq.h"
162 : : #include "pgstat.h"
163 : : #include "postmaster/interrupt.h"
164 : : #include "replication/logicallauncher.h"
165 : : #include "replication/logicalworker.h"
166 : : #include "replication/origin.h"
167 : : #include "replication/worker_internal.h"
168 : : #include "storage/ipc.h"
169 : : #include "storage/latch.h"
170 : : #include "storage/lmgr.h"
171 : : #include "storage/proc.h"
172 : : #include "tcop/tcopprot.h"
173 : : #include "utils/inval.h"
174 : : #include "utils/memutils.h"
175 : : #include "utils/syscache.h"
176 : : #include "utils/wait_event.h"
177 : :
178 : : #define PG_LOGICAL_APPLY_SHM_MAGIC 0x787ca067
179 : :
180 : : /*
181 : : * DSM keys for parallel apply worker. Unlike other parallel execution code,
182 : : * since we don't need to worry about DSM keys conflicting with plan_node_id we
183 : : * can use small integers.
184 : : */
185 : : #define PARALLEL_APPLY_KEY_SHARED 1
186 : : #define PARALLEL_APPLY_KEY_MQ 2
187 : : #define PARALLEL_APPLY_KEY_ERROR_QUEUE 3
188 : :
189 : : /* Queue size of DSM, 16 MB for now. */
190 : : #define DSM_QUEUE_SIZE (16 * 1024 * 1024)
191 : :
192 : : /*
193 : : * Error queue size of DSM. It is desirable to make it large enough that a
194 : : * typical ErrorResponse can be sent without blocking. That way, a worker that
195 : : * errors out can write the whole message into the queue and terminate without
196 : : * waiting for the user backend.
197 : : */
198 : : #define DSM_ERROR_QUEUE_SIZE (16 * 1024)
199 : :
200 : : /*
201 : : * There are three fields in each message received by the parallel apply
202 : : * worker: start_lsn, end_lsn and send_time. Because we have updated these
203 : : * statistics in the leader apply worker, we can ignore these fields in the
204 : : * parallel apply worker (see function LogicalRepApplyLoop).
205 : : */
206 : : #define SIZE_STATS_MESSAGE (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz))
207 : :
208 : : /*
209 : : * The type of session-level lock on a transaction being applied on a logical
210 : : * replication subscriber.
211 : : */
212 : : #define PARALLEL_APPLY_LOCK_STREAM 0
213 : : #define PARALLEL_APPLY_LOCK_XACT 1
214 : :
215 : : /*
216 : : * Hash table entry to map xid to the parallel apply worker state.
217 : : */
218 : : typedef struct ParallelApplyWorkerEntry
219 : : {
220 : : TransactionId xid; /* Hash key -- must be first */
221 : : ParallelApplyWorkerInfo *winfo;
222 : : } ParallelApplyWorkerEntry;
223 : :
224 : : /*
225 : : * A hash table used to cache the state of streaming transactions being applied
226 : : * by the parallel apply workers.
227 : : */
228 : : static HTAB *ParallelApplyTxnHash = NULL;
229 : :
230 : : /*
231 : : * A list (pool) of active parallel apply workers. The information for
232 : : * the new worker is added to the list after successfully launching it. The
233 : : * list entry is removed if there are already enough workers in the worker
234 : : * pool at the end of the transaction. For more information about the worker
235 : : * pool, see comments atop this file.
236 : : */
237 : : static List *ParallelApplyWorkerPool = NIL;
238 : :
239 : : /*
240 : : * Information shared between leader apply worker and parallel apply worker.
241 : : */
242 : : ParallelApplyWorkerShared *MyParallelShared = NULL;
243 : :
244 : : /*
245 : : * Is there a message sent by a parallel apply worker that the leader apply
246 : : * worker needs to receive?
247 : : */
248 : : volatile sig_atomic_t ParallelApplyMessagePending = false;
249 : :
250 : : /*
251 : : * Cache the parallel apply worker information required for applying the
252 : : * current streaming transaction. It is used to save the cost of searching the
253 : : * hash table when applying the changes between STREAM_START and STREAM_STOP.
254 : : */
255 : : static ParallelApplyWorkerInfo *stream_apply_worker = NULL;
256 : :
257 : : /* A list to maintain subtransactions, if any. */
258 : : static List *subxactlist = NIL;
259 : :
260 : : static void pa_free_worker_info(ParallelApplyWorkerInfo *winfo);
261 : : static ParallelTransState pa_get_xact_state(ParallelApplyWorkerShared *wshared);
262 : : static PartialFileSetState pa_get_fileset_state(void);
263 : :
264 : : /*
265 : : * Returns true if it is OK to start a parallel apply worker, false otherwise.
266 : : */
267 : : static bool
1326 akapila@postgresql.o 268 :CBC 86 : pa_can_start(void)
269 : : {
270 : : /* Only leader apply workers can start parallel apply workers. */
271 [ + + ]: 86 : if (!am_leader_apply_worker())
272 : 29 : return false;
273 : :
274 : : /*
275 : : * It is good to check for any change in the subscription parameter to
276 : : * avoid the case where for a very long time the change doesn't get
277 : : * reflected. This can happen when there is a constant flow of streaming
278 : : * transactions that are handled by parallel apply workers.
279 : : *
280 : : * It is better to do it before the below checks so that the latest values
281 : : * of subscription can be used for the checks.
282 : : */
283 : 57 : maybe_reread_subscription();
284 : :
285 : : /*
286 : : * Don't start a new parallel apply worker if the subscription is not
287 : : * using parallel streaming mode, or if the publisher does not support
288 : : * parallel apply.
289 : : */
290 [ + + ]: 57 : if (!MyLogicalRepWorker->parallel_apply)
291 : 28 : return false;
292 : :
293 : : /*
294 : : * Don't start a new parallel worker if user has set skiplsn as it's
295 : : * possible that they want to skip the streaming transaction. For
296 : : * streaming transactions, we need to serialize the transaction to a file
297 : : * so that we can get the last LSN of the transaction to judge whether to
298 : : * skip before starting to apply the change.
299 : : *
300 : : * One might think that we could allow parallelism if the first lsn of the
301 : : * transaction is greater than skiplsn, but we don't send it with the
302 : : * STREAM START message, and it doesn't seem worth sending the extra eight
303 : : * bytes with the STREAM START to enable parallelism for this case.
304 : : */
294 alvherre@kurilemu.de 305 [ - + ]: 29 : if (XLogRecPtrIsValid(MySubscription->skiplsn))
1326 akapila@postgresql.o 306 :UBC 0 : return false;
307 : :
308 : : /*
309 : : * For streaming transactions that are being applied using a parallel
310 : : * apply worker, we cannot decide whether to apply the change for a
311 : : * relation that is not in the READY state (see
312 : : * should_apply_changes_for_rel) as we won't know the finish LSN of the
313 : : * transaction by that time. So, we don't start the new parallel apply
314 : : * worker in this case.
315 : : */
1326 akapila@postgresql.o 316 [ - + ]:CBC 29 : if (!AllTablesyncsReady())
1326 akapila@postgresql.o 317 :UBC 0 : return false;
318 : :
1326 akapila@postgresql.o 319 :CBC 29 : return true;
320 : : }
321 : :
322 : : /*
323 : : * Set up a dynamic shared memory segment.
324 : : *
325 : : * We set up a control region that contains a fixed-size worker info
326 : : * (ParallelApplyWorkerShared), a message queue, and an error queue.
327 : : *
328 : : * Returns true on success, false on failure.
329 : : */
330 : : static bool
331 : 12 : pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
332 : : {
333 : : shm_toc_estimator e;
334 : : Size segsize;
335 : : dsm_segment *seg;
336 : : shm_toc *toc;
337 : : ParallelApplyWorkerShared *shared;
338 : : shm_mq *mq;
339 : 12 : Size queue_size = DSM_QUEUE_SIZE;
340 : 12 : Size error_queue_size = DSM_ERROR_QUEUE_SIZE;
341 : :
342 : : /*
343 : : * Estimate how much shared memory we need.
344 : : *
345 : : * Because the TOC machinery may choose to insert padding of oddly-sized
346 : : * requests, we must estimate each chunk separately.
347 : : *
348 : : * We need one key to register the location of the header, and two other
349 : : * keys to track the locations of the message queue and the error message
350 : : * queue.
351 : : */
352 : 12 : shm_toc_initialize_estimator(&e);
353 : 12 : shm_toc_estimate_chunk(&e, sizeof(ParallelApplyWorkerShared));
354 : 12 : shm_toc_estimate_chunk(&e, queue_size);
355 : 12 : shm_toc_estimate_chunk(&e, error_queue_size);
356 : :
357 : 12 : shm_toc_estimate_keys(&e, 3);
358 : 12 : segsize = shm_toc_estimate(&e);
359 : :
360 : : /* Create the shared memory segment and establish a table of contents. */
361 : 12 : seg = dsm_create(shm_toc_estimate(&e), 0);
362 [ - + ]: 12 : if (!seg)
1326 akapila@postgresql.o 363 :UBC 0 : return false;
364 : :
1326 akapila@postgresql.o 365 :CBC 12 : toc = shm_toc_create(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg),
366 : : segsize);
367 : :
368 : : /* Set up the header region. */
369 : 12 : shared = shm_toc_allocate(toc, sizeof(ParallelApplyWorkerShared));
370 : 12 : SpinLockInit(&shared->mutex);
371 : :
372 : 12 : shared->xact_state = PARALLEL_TRANS_UNKNOWN;
373 : 12 : pg_atomic_init_u32(&(shared->pending_stream_count), 0);
374 : 12 : shared->last_commit_end = InvalidXLogRecPtr;
375 : 12 : shared->fileset_state = FS_EMPTY;
376 : :
377 : 12 : shm_toc_insert(toc, PARALLEL_APPLY_KEY_SHARED, shared);
378 : :
379 : : /* Set up message queue for the worker. */
380 : 12 : mq = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
381 : 12 : shm_toc_insert(toc, PARALLEL_APPLY_KEY_MQ, mq);
382 : 12 : shm_mq_set_sender(mq, MyProc);
383 : :
384 : : /* Attach the queue. */
385 : 12 : winfo->mq_handle = shm_mq_attach(mq, seg, NULL);
386 : :
387 : : /* Set up error queue for the worker. */
388 : 12 : mq = shm_mq_create(shm_toc_allocate(toc, error_queue_size),
389 : : error_queue_size);
390 : 12 : shm_toc_insert(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, mq);
391 : 12 : shm_mq_set_receiver(mq, MyProc);
392 : :
393 : : /* Attach the queue. */
394 : 12 : winfo->error_mq_handle = shm_mq_attach(mq, seg, NULL);
395 : :
396 : : /* Return results to caller. */
397 : 12 : winfo->dsm_seg = seg;
398 : 12 : winfo->shared = shared;
399 : :
400 : 12 : return true;
401 : : }
402 : :
403 : : /*
404 : : * Try to get a parallel apply worker from the pool. If none is available then
405 : : * start a new one.
406 : : */
407 : : static ParallelApplyWorkerInfo *
408 : 29 : pa_launch_parallel_worker(void)
409 : : {
410 : : MemoryContext oldcontext;
411 : : bool launched;
412 : : ParallelApplyWorkerInfo *winfo;
413 : : ListCell *lc;
414 : :
415 : : /* Try to get an available parallel apply worker from the worker pool. */
416 [ + + + + : 31 : foreach(lc, ParallelApplyWorkerPool)
+ + ]
417 : : {
418 : 19 : winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
419 : :
420 [ + + ]: 19 : if (!winfo->in_use)
421 : 17 : return winfo;
422 : : }
423 : :
424 : : /*
425 : : * Start a new parallel apply worker.
426 : : *
427 : : * The worker info can be used for the lifetime of the worker process, so
428 : : * create it in a permanent context.
429 : : */
430 : 12 : oldcontext = MemoryContextSwitchTo(ApplyContext);
431 : :
260 michael@paquier.xyz 432 : 12 : winfo = palloc0_object(ParallelApplyWorkerInfo);
433 : :
434 : : /* Setup shared memory. */
1326 akapila@postgresql.o 435 [ - + ]: 12 : if (!pa_setup_dsm(winfo))
436 : : {
1326 akapila@postgresql.o 437 :UBC 0 : MemoryContextSwitchTo(oldcontext);
438 : 0 : pfree(winfo);
439 : 0 : return NULL;
440 : : }
441 : :
1109 akapila@postgresql.o 442 :CBC 12 : launched = logicalrep_worker_launch(WORKERTYPE_PARALLEL_APPLY,
443 : 12 : MyLogicalRepWorker->dbid,
1326 444 : 12 : MySubscription->oid,
445 : 12 : MySubscription->name,
446 : 12 : MyLogicalRepWorker->userid,
447 : : InvalidOid,
448 : : dsm_segment_handle(winfo->dsm_seg),
449 : : false);
450 : :
451 [ + - ]: 12 : if (launched)
452 : : {
453 : 12 : ParallelApplyWorkerPool = lappend(ParallelApplyWorkerPool, winfo);
454 : : }
455 : : else
456 : : {
1326 akapila@postgresql.o 457 :UBC 0 : pa_free_worker_info(winfo);
458 : 0 : winfo = NULL;
459 : : }
460 : :
1326 akapila@postgresql.o 461 :CBC 12 : MemoryContextSwitchTo(oldcontext);
462 : :
463 : 12 : return winfo;
464 : : }
465 : :
466 : : /*
467 : : * Allocate a parallel apply worker that will be used for the specified xid.
468 : : *
469 : : * We first try to get an available worker from the pool, if any and then try
470 : : * to launch a new worker. On successful allocation, remember the worker
471 : : * information in the hash table so that we can get it later for processing the
472 : : * streaming changes.
473 : : */
474 : : void
475 : 86 : pa_allocate_worker(TransactionId xid)
476 : : {
477 : : bool found;
478 : 86 : ParallelApplyWorkerInfo *winfo = NULL;
479 : : ParallelApplyWorkerEntry *entry;
480 : :
481 [ + + ]: 86 : if (!pa_can_start())
482 : 57 : return;
483 : :
1322 484 : 29 : winfo = pa_launch_parallel_worker();
485 [ - + ]: 29 : if (!winfo)
1322 akapila@postgresql.o 486 :UBC 0 : return;
487 : :
488 : : /* First time through, initialize parallel apply worker state hashtable. */
1326 akapila@postgresql.o 489 [ + + ]:CBC 29 : if (!ParallelApplyTxnHash)
490 : : {
491 : : HASHCTL ctl;
492 : :
493 [ + - + - : 99 : MemSet(&ctl, 0, sizeof(ctl));
+ - + - +
+ ]
494 : 9 : ctl.keysize = sizeof(TransactionId);
495 : 9 : ctl.entrysize = sizeof(ParallelApplyWorkerEntry);
496 : 9 : ctl.hcxt = ApplyContext;
497 : :
498 : 9 : ParallelApplyTxnHash = hash_create("logical replication parallel apply workers hash",
499 : : 16, &ctl,
500 : : HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
501 : : }
502 : :
503 : : /* Create an entry for the requested transaction. */
504 : 29 : entry = hash_search(ParallelApplyTxnHash, &xid, HASH_ENTER, &found);
505 [ - + ]: 29 : if (found)
1326 akapila@postgresql.o 506 [ # # ]:UBC 0 : elog(ERROR, "hash table corrupted");
507 : :
508 : : /* Update the transaction information in shared memory. */
1326 akapila@postgresql.o 509 :CBC 29 : SpinLockAcquire(&winfo->shared->mutex);
510 : 29 : winfo->shared->xact_state = PARALLEL_TRANS_UNKNOWN;
511 : 29 : winfo->shared->xid = xid;
512 : 29 : SpinLockRelease(&winfo->shared->mutex);
513 : :
514 : 29 : winfo->in_use = true;
515 : 29 : winfo->serialize_changes = false;
516 : 29 : entry->winfo = winfo;
517 : : }
518 : :
519 : : /*
520 : : * Find the assigned worker for the given transaction, if any.
521 : : */
522 : : ParallelApplyWorkerInfo *
523 : 277892 : pa_find_worker(TransactionId xid)
524 : : {
525 : : bool found;
526 : : ParallelApplyWorkerEntry *entry;
527 : :
528 [ + + ]: 277892 : if (!TransactionIdIsValid(xid))
529 : 105694 : return NULL;
530 : :
531 [ + + ]: 172198 : if (!ParallelApplyTxnHash)
532 : 103241 : return NULL;
533 : :
534 : : /* Return the cached parallel apply worker if valid. */
535 [ + + ]: 68957 : if (stream_apply_worker)
536 : 68661 : return stream_apply_worker;
537 : :
538 : : /* Find an entry for the requested transaction. */
539 : 296 : entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
540 [ + - ]: 296 : if (found)
541 : : {
542 : : /* The worker must not have exited. */
543 [ - + ]: 296 : Assert(entry->winfo->in_use);
544 : 296 : return entry->winfo;
545 : : }
546 : :
1326 akapila@postgresql.o 547 :UBC 0 : return NULL;
548 : : }
549 : :
550 : : /*
551 : : * Makes the worker available for reuse.
552 : : *
553 : : * This removes the parallel apply worker entry from the hash table so that it
554 : : * can't be used. If there are enough workers in the pool, it stops the worker
555 : : * and frees the corresponding info. Otherwise it just marks the worker as
556 : : * available for reuse.
557 : : *
558 : : * For more information about the worker pool, see comments atop this file.
559 : : */
560 : : static void
1326 akapila@postgresql.o 561 :CBC 25 : pa_free_worker(ParallelApplyWorkerInfo *winfo)
562 : : {
563 [ - + ]: 25 : Assert(!am_parallel_apply_worker());
564 [ - + ]: 25 : Assert(winfo->in_use);
565 [ - + ]: 25 : Assert(pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_FINISHED);
566 : :
567 [ - + ]: 25 : if (!hash_search(ParallelApplyTxnHash, &winfo->shared->xid, HASH_REMOVE, NULL))
1326 akapila@postgresql.o 568 [ # # ]:UBC 0 : elog(ERROR, "hash table corrupted");
569 : :
570 : : /*
571 : : * Stop the worker if there are enough workers in the pool.
572 : : *
573 : : * XXX Additionally, we also stop the worker if the leader apply worker
574 : : * serialize part of the transaction data due to a send timeout. This is
575 : : * because the message could be partially written to the queue and there
576 : : * is no way to clean the queue other than resending the message until it
577 : : * succeeds. Instead of trying to send the data which anyway would have
578 : : * been serialized and then letting the parallel apply worker deal with
579 : : * the spurious message, we stop the worker.
580 : : */
1326 akapila@postgresql.o 581 [ + + ]:CBC 25 : if (winfo->serialize_changes ||
582 : 21 : list_length(ParallelApplyWorkerPool) >
583 [ + + ]: 21 : (max_parallel_apply_workers_per_subscription / 2))
584 : : {
1206 585 : 5 : logicalrep_pa_worker_stop(winfo);
1326 586 : 5 : pa_free_worker_info(winfo);
587 : :
588 : 5 : return;
589 : : }
590 : :
591 : 20 : winfo->in_use = false;
592 : 20 : winfo->serialize_changes = false;
593 : : }
594 : :
595 : : /*
596 : : * Free the parallel apply worker information and unlink the files with
597 : : * serialized changes if any.
598 : : */
599 : : static void
600 : 5 : pa_free_worker_info(ParallelApplyWorkerInfo *winfo)
601 : : {
602 [ - + ]: 5 : Assert(winfo);
603 : :
604 [ + - ]: 5 : if (winfo->mq_handle)
605 : 5 : shm_mq_detach(winfo->mq_handle);
606 : :
607 [ - + ]: 5 : if (winfo->error_mq_handle)
1326 akapila@postgresql.o 608 :UBC 0 : shm_mq_detach(winfo->error_mq_handle);
609 : :
610 : : /* Unlink the files with serialized changes. */
1326 akapila@postgresql.o 611 [ + + ]:CBC 5 : if (winfo->serialize_changes)
612 : 4 : stream_cleanup_files(MyLogicalRepWorker->subid, winfo->shared->xid);
613 : :
614 [ + - ]: 5 : if (winfo->dsm_seg)
615 : 5 : dsm_detach(winfo->dsm_seg);
616 : :
617 : : /* Remove from the worker pool. */
618 : 5 : ParallelApplyWorkerPool = list_delete_ptr(ParallelApplyWorkerPool, winfo);
619 : :
620 : 5 : pfree(winfo);
621 : 5 : }
622 : :
623 : : /*
624 : : * Detach the error queue for all parallel apply workers.
625 : : */
626 : : void
627 : 380 : pa_detach_all_error_mq(void)
628 : : {
629 : : ListCell *lc;
630 : :
631 [ + + + + : 387 : foreach(lc, ParallelApplyWorkerPool)
+ + ]
632 : : {
633 : 7 : ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
634 : :
1206 635 [ + - ]: 7 : if (winfo->error_mq_handle)
636 : : {
637 : 7 : shm_mq_detach(winfo->error_mq_handle);
638 : 7 : winfo->error_mq_handle = NULL;
639 : : }
640 : : }
1326 641 : 380 : }
642 : :
643 : : /*
644 : : * Check if there are any pending spooled messages.
645 : : */
646 : : static bool
267 nathan@postgresql.or 647 : 16 : pa_has_spooled_message_pending(void)
648 : : {
649 : : PartialFileSetState fileset_state;
650 : :
1326 akapila@postgresql.o 651 : 16 : fileset_state = pa_get_fileset_state();
652 : :
653 : 16 : return (fileset_state != FS_EMPTY);
654 : : }
655 : :
656 : : /*
657 : : * Replay the spooled messages once the leader apply worker has finished
658 : : * serializing changes to the file.
659 : : *
660 : : * Returns false if there aren't any pending spooled messages, true otherwise.
661 : : */
662 : : static bool
663 : 71 : pa_process_spooled_messages_if_required(void)
664 : : {
665 : : PartialFileSetState fileset_state;
666 : :
667 : 71 : fileset_state = pa_get_fileset_state();
668 : :
669 [ + + ]: 71 : if (fileset_state == FS_EMPTY)
670 : 63 : return false;
671 : :
672 : : /*
673 : : * If the leader apply worker is busy serializing the partial changes then
674 : : * acquire the stream lock now and wait for the leader worker to finish
675 : : * serializing the changes. Otherwise, the parallel apply worker won't get
676 : : * a chance to receive a STREAM_STOP (and acquire the stream lock) until
677 : : * the leader had serialized all changes which can lead to undetected
678 : : * deadlock.
679 : : *
680 : : * Note that the fileset state can be FS_SERIALIZE_DONE once the leader
681 : : * worker has finished serializing the changes.
682 : : */
683 [ - + ]: 8 : if (fileset_state == FS_SERIALIZE_IN_PROGRESS)
684 : : {
1326 akapila@postgresql.o 685 :UBC 0 : pa_lock_stream(MyParallelShared->xid, AccessShareLock);
686 : 0 : pa_unlock_stream(MyParallelShared->xid, AccessShareLock);
687 : :
688 : 0 : fileset_state = pa_get_fileset_state();
689 : : }
690 : :
691 : : /*
692 : : * We cannot read the file immediately after the leader has serialized all
693 : : * changes to the file because there may still be messages in the memory
694 : : * queue. We will apply all spooled messages the next time we call this
695 : : * function and that will ensure there are no messages left in the memory
696 : : * queue.
697 : : */
1326 akapila@postgresql.o 698 [ + + ]:CBC 8 : if (fileset_state == FS_SERIALIZE_DONE)
699 : : {
700 : 4 : pa_set_fileset_state(MyParallelShared, FS_READY);
701 : : }
702 [ + - ]: 4 : else if (fileset_state == FS_READY)
703 : : {
704 : 4 : apply_spooled_messages(&MyParallelShared->fileset,
705 : 4 : MyParallelShared->xid,
706 : : InvalidXLogRecPtr);
707 : 4 : pa_set_fileset_state(MyParallelShared, FS_EMPTY);
708 : : }
709 : :
710 : 8 : return true;
711 : : }
712 : :
713 : : /*
714 : : * Interrupt handler for main loop of parallel apply worker.
715 : : */
716 : : static void
717 : 58975 : ProcessParallelApplyInterrupts(void)
718 : : {
719 [ + + ]: 58975 : CHECK_FOR_INTERRUPTS();
720 : :
721 [ + + ]: 58971 : if (ShutdownRequestPending)
722 : : {
723 [ + - ]: 5 : ereport(LOG,
724 : : (errmsg("logical replication parallel apply worker for subscription \"%s\" has finished",
725 : : MySubscription->name)));
726 : :
727 : 5 : proc_exit(0);
728 : : }
729 : :
730 [ + + ]: 58966 : if (ConfigReloadPending)
731 : : {
732 : 4 : ConfigReloadPending = false;
733 : 4 : ProcessConfigFile(PGC_SIGHUP);
734 : : }
735 : 58966 : }
736 : :
737 : : /* Parallel apply worker main loop. */
738 : : static void
739 : 12 : LogicalParallelApplyLoop(shm_mq_handle *mqh)
740 : : {
741 : : shm_mq_result shmq_res;
742 : : ErrorContextCallback errcallback;
743 : 12 : MemoryContext oldcxt = CurrentMemoryContext;
744 : :
745 : : /*
746 : : * Init the ApplyMessageContext which we clean up after each replication
747 : : * protocol message.
748 : : */
749 : 12 : ApplyMessageContext = AllocSetContextCreate(ApplyContext,
750 : : "ApplyMessageContext",
751 : : ALLOCSET_DEFAULT_SIZES);
752 : :
753 : : /*
754 : : * Push apply error context callback. Fields will be filled while applying
755 : : * a change.
756 : : */
757 : 12 : errcallback.callback = apply_error_callback;
758 : 12 : errcallback.previous = error_context_stack;
759 : 12 : error_context_stack = &errcallback;
760 : :
761 : : for (;;)
762 : 58963 : {
763 : : void *data;
764 : : Size len;
765 : :
766 : 58975 : ProcessParallelApplyInterrupts();
767 : :
768 : : /* Ensure we are reading the data into our memory context. */
769 : 58966 : MemoryContextSwitchTo(ApplyMessageContext);
770 : :
771 : 58966 : shmq_res = shm_mq_receive(mqh, &len, &data, true);
772 : :
773 [ + + ]: 58966 : if (shmq_res == SHM_MQ_SUCCESS)
774 : : {
775 : : StringInfoData s;
776 : : int c;
777 : :
778 [ - + ]: 58895 : if (len == 0)
1326 akapila@postgresql.o 779 [ # # ]:UBC 0 : elog(ERROR, "invalid message length");
780 : :
1036 drowley@postgresql.o 781 :CBC 58895 : initReadOnlyStringInfo(&s, data, len);
782 : :
783 : : /*
784 : : * The first byte of messages sent from leader apply worker to
785 : : * parallel apply workers can only be PqReplMsg_WALData.
786 : : */
1326 akapila@postgresql.o 787 : 58895 : c = pq_getmsgbyte(&s);
386 nathan@postgresql.or 788 [ - + ]: 58895 : if (c != PqReplMsg_WALData)
1326 akapila@postgresql.o 789 [ # # ]:UBC 0 : elog(ERROR, "unexpected message \"%c\"", c);
790 : :
791 : : /*
792 : : * Ignore statistics fields that have been updated by the leader
793 : : * apply worker.
794 : : *
795 : : * XXX We can avoid sending the statistics fields from the leader
796 : : * apply worker but for that, it needs to rebuild the entire
797 : : * message by removing these fields which could be more work than
798 : : * simply ignoring these fields in the parallel apply worker.
799 : : */
1326 akapila@postgresql.o 800 :CBC 58895 : s.cursor += SIZE_STATS_MESSAGE;
801 : :
802 : 58895 : apply_dispatch(&s);
803 : : }
804 [ + - ]: 71 : else if (shmq_res == SHM_MQ_WOULD_BLOCK)
805 : : {
806 : : /* Replay the changes from the file, if any. */
807 [ + + ]: 71 : if (!pa_process_spooled_messages_if_required())
808 : : {
809 : : int rc;
810 : :
811 : : /* Wait for more work. */
812 : 63 : rc = WaitLatch(MyLatch,
813 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
814 : : 1000L,
815 : : WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN);
816 : :
817 [ + + ]: 63 : if (rc & WL_LATCH_SET)
818 : 60 : ResetLatch(MyLatch);
819 : :
820 : : /*
821 : : * Force stats reporting to avoid long delays. There can be
822 : : * long idle gaps before the leader assigns the next
823 : : * transaction, and the only opportunity to report stats
824 : : * during such gaps is here.
825 : : */
129 826 [ + + + - ]: 63 : if ((rc & WL_TIMEOUT) && !IsTransactionState())
827 : 3 : pgstat_report_stat(true);
828 : : }
829 : : }
830 : : else
831 : : {
1326 akapila@postgresql.o 832 [ # # ]:UBC 0 : Assert(shmq_res == SHM_MQ_DETACHED);
833 : :
834 [ # # ]: 0 : ereport(ERROR,
835 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
836 : : errmsg("lost connection to the logical replication apply worker")));
837 : : }
838 : :
1326 akapila@postgresql.o 839 :CBC 58963 : MemoryContextReset(ApplyMessageContext);
840 : 58963 : MemoryContextSwitchTo(oldcxt);
841 : : }
842 : :
843 : : /* Pop the error context stack. */
844 : : error_context_stack = errcallback.previous;
845 : :
846 : : MemoryContextSwitchTo(oldcxt);
847 : : }
848 : :
849 : : /*
850 : : * Make sure the leader apply worker tries to read from our error queue one more
851 : : * time. This guards against the case where we exit uncleanly without sending
852 : : * an ErrorResponse, for example because some code calls proc_exit directly.
853 : : *
854 : : * Also explicitly detach from dsm segment to invoke on_dsm_detach callbacks,
855 : : * if any. See ParallelWorkerShutdown for details.
856 : : */
857 : : static void
858 : 12 : pa_shutdown(int code, Datum arg)
859 : : {
1317 860 : 12 : SendProcSignal(MyLogicalRepWorker->leader_pid,
861 : : PROCSIG_PARALLEL_APPLY_MESSAGE,
862 : : INVALID_PROC_NUMBER);
863 : :
1326 864 : 12 : dsm_detach((dsm_segment *) DatumGetPointer(arg));
865 : 12 : }
866 : :
867 : : /*
868 : : * Parallel apply worker entry point.
869 : : */
870 : : void
871 : 12 : ParallelApplyWorkerMain(Datum main_arg)
872 : : {
873 : : ParallelApplyWorkerShared *shared;
874 : : dsm_handle handle;
875 : : dsm_segment *seg;
876 : : shm_toc *toc;
877 : : shm_mq *mq;
878 : : shm_mq_handle *mqh;
879 : : shm_mq_handle *error_mqh;
880 : : ReplOriginId originid;
881 : 12 : int worker_slot = DatumGetInt32(main_arg);
882 : : char originname[NAMEDATALEN];
883 : :
1212 884 : 12 : InitializingApplyWorker = true;
885 : :
886 : : /*
887 : : * Setup signal handling.
888 : : *
889 : : * Note: We intentionally used SIGUSR2 to trigger a graceful shutdown
890 : : * initiated by the leader apply worker. This helps to differentiate it
891 : : * from the case where we abort the current transaction and exit on
892 : : * receiving SIGTERM.
893 : : */
1326 894 : 12 : pqsignal(SIGHUP, SignalHandlerForConfigReload);
337 895 : 12 : pqsignal(SIGUSR2, SignalHandlerForShutdownRequest);
1326 896 : 12 : BackgroundWorkerUnblockSignals();
897 : :
898 : : /*
899 : : * Attach to the dynamic shared memory segment for the parallel apply, and
900 : : * find its table of contents.
901 : : *
902 : : * Like parallel query, we don't need resource owner by this time. See
903 : : * ParallelWorkerMain.
904 : : */
905 : 12 : memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
906 : 12 : seg = dsm_attach(handle);
907 [ - + ]: 12 : if (!seg)
1326 akapila@postgresql.o 908 [ # # ]:UBC 0 : ereport(ERROR,
909 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
910 : : errmsg("could not map dynamic shared memory segment")));
911 : :
1326 akapila@postgresql.o 912 :CBC 12 : toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
913 [ - + ]: 12 : if (!toc)
1326 akapila@postgresql.o 914 [ # # ]:UBC 0 : ereport(ERROR,
915 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
916 : : errmsg("invalid magic number in dynamic shared memory segment")));
917 : :
918 : : /* Look up the shared information. */
1326 akapila@postgresql.o 919 :CBC 12 : shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
920 : 12 : MyParallelShared = shared;
921 : :
922 : : /*
923 : : * Attach to the message queue.
924 : : */
925 : 12 : mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
926 : 12 : shm_mq_set_receiver(mq, MyProc);
927 : 12 : mqh = shm_mq_attach(mq, seg, NULL);
928 : :
929 : : /*
930 : : * Primary initialization is complete. Now, we can attach to our slot.
931 : : * This is to ensure that the leader apply worker does not write data to
932 : : * the uninitialized memory queue.
933 : : */
934 : 12 : logicalrep_worker_attach(worker_slot);
935 : :
936 : : /*
937 : : * Register the shutdown callback after we are attached to the worker
938 : : * slot. This is to ensure that MyLogicalRepWorker remains valid when this
939 : : * callback is invoked.
940 : : */
1206 941 : 12 : before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
942 : :
1326 943 : 12 : SpinLockAcquire(&MyParallelShared->mutex);
944 : 12 : MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
945 : 12 : MyParallelShared->logicalrep_worker_slot_no = worker_slot;
946 : 12 : SpinLockRelease(&MyParallelShared->mutex);
947 : :
948 : : /*
949 : : * Attach to the error queue.
950 : : */
951 : 12 : mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, false);
952 : 12 : shm_mq_set_sender(mq, MyProc);
953 : 12 : error_mqh = shm_mq_attach(mq, seg, NULL);
954 : :
955 : 12 : pq_redirect_to_shm_mq(seg, error_mqh);
1317 956 : 12 : pq_set_parallel_leader(MyLogicalRepWorker->leader_pid,
957 : : INVALID_PROC_NUMBER);
958 : :
1326 959 : 12 : MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
960 : 12 : MyLogicalRepWorker->reply_time = 0;
961 : :
1120 962 : 12 : InitializeLogRepWorker();
963 : :
1212 964 : 12 : InitializingApplyWorker = false;
965 : :
966 : : /* Setup replication origin tracking. */
1326 967 : 12 : StartTransactionCommand();
968 : 12 : ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
969 : : originname, sizeof(originname));
970 : 12 : originid = replorigin_by_name(originname, false);
971 : :
972 : : /*
973 : : * The parallel apply worker doesn't need to monopolize this replication
974 : : * origin which was already acquired by its leader process.
975 : : */
1317 976 : 12 : replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
211 msawada@postgresql.o 977 : 12 : replorigin_xact_state.origin = originid;
1326 akapila@postgresql.o 978 : 12 : CommitTransactionCommand();
979 : :
980 : : /*
981 : : * Setup callback for syscache so that we know when something changes in
982 : : * the subscription relation state.
983 : : */
984 : 12 : CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
985 : : InvalidateSyncingRelStates,
986 : : (Datum) 0);
987 : :
988 : 12 : set_apply_error_context_origin(originname);
989 : :
990 : 12 : LogicalParallelApplyLoop(mqh);
991 : :
992 : : /*
993 : : * The parallel apply worker must not get here because the parallel apply
994 : : * worker will only stop when it receives a SIGTERM or SIGUSR2 from the
995 : : * leader, or SIGINT from itself, or when there is an error. None of these
996 : : * cases will allow the code to reach here.
997 : : */
1326 akapila@postgresql.o 998 :UBC 0 : Assert(false);
999 : : }
1000 : :
1001 : : /*
1002 : : * Handle receipt of an interrupt indicating a parallel apply worker message.
1003 : : *
1004 : : * Note: this is called within a signal handler! All we can do is set a flag
1005 : : * that will cause the next CHECK_FOR_INTERRUPTS() to invoke
1006 : : * ProcessParallelApplyMessages().
1007 : : */
1008 : : void
1326 akapila@postgresql.o 1009 :CBC 15 : HandleParallelApplyMessageInterrupt(void)
1010 : : {
1011 : 15 : InterruptPending = true;
1012 : 15 : ParallelApplyMessagePending = true;
1013 : : /* latch will be set by procsignal_sigusr1_handler */
1014 : 15 : }
1015 : :
1016 : : /*
1017 : : * Process a single protocol message received from a single parallel apply
1018 : : * worker.
1019 : : */
1020 : : static void
540 heikki.linnakangas@i 1021 : 3 : ProcessParallelApplyMessage(StringInfo msg)
1022 : : {
1023 : : char msgtype;
1024 : :
1326 akapila@postgresql.o 1025 : 3 : msgtype = pq_getmsgbyte(msg);
1026 : :
1027 [ + - - ]: 3 : switch (msgtype)
1028 : : {
367 nathan@postgresql.or 1029 : 3 : case PqMsg_ErrorResponse:
1030 : : {
1031 : : ErrorData edata;
1032 : :
1033 : : /* Parse ErrorResponse. */
1326 akapila@postgresql.o 1034 : 3 : pq_parse_errornotice(msg, &edata);
1035 : :
1036 : : /*
1037 : : * If desired, add a context line to show that this is a
1038 : : * message propagated from a parallel apply worker. Otherwise,
1039 : : * it can sometimes be confusing to understand what actually
1040 : : * happened.
1041 : : */
1042 [ + - ]: 3 : if (edata.context)
1043 : 3 : edata.context = psprintf("%s\n%s", edata.context,
1044 : : _("logical replication parallel apply worker"));
1045 : : else
1326 akapila@postgresql.o 1046 :UBC 0 : edata.context = pstrdup(_("logical replication parallel apply worker"));
1047 : :
1048 : : /*
1049 : : * Context beyond that should use the error context callbacks
1050 : : * that were in effect in LogicalRepApplyLoop().
1051 : : */
1326 akapila@postgresql.o 1052 :CBC 3 : error_context_stack = apply_error_context_stack;
1053 : :
1054 : : /*
1055 : : * The actual error must have been reported by the parallel
1056 : : * apply worker.
1057 : : */
1058 [ + - ]: 3 : ereport(ERROR,
1059 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1060 : : errmsg("logical replication parallel apply worker exited due to error"),
1061 : : errcontext("%s", edata.context)));
1062 : : }
1063 : :
1064 : : /*
1065 : : * Don't need to do anything about NoticeResponse and
1066 : : * NotificationResponse as the logical replication worker doesn't
1067 : : * need to send messages to the client.
1068 : : */
367 nathan@postgresql.or 1069 :UBC 0 : case PqMsg_NoticeResponse:
1070 : : case PqMsg_NotificationResponse:
1326 akapila@postgresql.o 1071 : 0 : break;
1072 : :
1073 : 0 : default:
1074 [ # # ]: 0 : elog(ERROR, "unrecognized message type received from logical replication parallel apply worker: %c (message length %d bytes)",
1075 : : msgtype, msg->len);
1076 : : }
1077 : 0 : }
1078 : :
1079 : : /*
1080 : : * Handle any queued protocol messages received from parallel apply workers.
1081 : : */
1082 : : void
540 heikki.linnakangas@i 1083 :CBC 8 : ProcessParallelApplyMessages(void)
1084 : : {
1085 : : ListCell *lc;
1086 : : MemoryContext oldcontext;
1087 : :
1088 : : static MemoryContext hpam_context = NULL;
1089 : :
1090 : : /*
1091 : : * This is invoked from ProcessInterrupts(), and since some of the
1092 : : * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
1093 : : * for recursive calls if more signals are received while this runs. It's
1094 : : * unclear that recursive entry would be safe, and it doesn't seem useful
1095 : : * even if it is safe, so let's block interrupts until done.
1096 : : */
1326 akapila@postgresql.o 1097 : 8 : HOLD_INTERRUPTS();
1098 : :
1099 : : /*
1100 : : * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
1101 : : * don't want to risk leaking data into long-lived contexts, so let's do
1102 : : * our work here in a private context that we can reset on each use.
1103 : : */
1104 [ + + ]: 8 : if (!hpam_context) /* first time through? */
1105 : 7 : hpam_context = AllocSetContextCreate(TopMemoryContext,
1106 : : "ProcessParallelApplyMessages",
1107 : : ALLOCSET_DEFAULT_SIZES);
1108 : : else
1109 : 1 : MemoryContextReset(hpam_context);
1110 : :
1111 : 8 : oldcontext = MemoryContextSwitchTo(hpam_context);
1112 : :
1113 : 8 : ParallelApplyMessagePending = false;
1114 : :
1115 [ + - + + : 14 : foreach(lc, ParallelApplyWorkerPool)
+ + ]
1116 : : {
1117 : : shm_mq_result res;
1118 : : Size nbytes;
1119 : : void *data;
1120 : 9 : ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
1121 : :
1122 : : /*
1123 : : * The leader will detach from the error queue and set it to NULL
1124 : : * before preparing to stop all parallel apply workers, so we don't
1125 : : * need to handle error messages anymore. See
1126 : : * logicalrep_worker_detach.
1127 : : */
1128 [ + + ]: 9 : if (!winfo->error_mq_handle)
1129 : 6 : continue;
1130 : :
1131 : 4 : res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
1132 : :
1133 [ + + ]: 4 : if (res == SHM_MQ_WOULD_BLOCK)
1134 : 1 : continue;
1135 [ + - ]: 3 : else if (res == SHM_MQ_SUCCESS)
1136 : : {
1137 : : StringInfoData msg;
1138 : :
1139 : 3 : initStringInfo(&msg);
1140 : 3 : appendBinaryStringInfo(&msg, data, nbytes);
540 heikki.linnakangas@i 1141 : 3 : ProcessParallelApplyMessage(&msg);
1326 akapila@postgresql.o 1142 :UBC 0 : pfree(msg.data);
1143 : : }
1144 : : else
1145 [ # # ]: 0 : ereport(ERROR,
1146 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1147 : : errmsg("lost connection to the logical replication parallel apply worker")));
1148 : : }
1149 : :
1326 akapila@postgresql.o 1150 :CBC 5 : MemoryContextSwitchTo(oldcontext);
1151 : :
1152 : : /* Might as well clear the context on our way out */
1153 : 5 : MemoryContextReset(hpam_context);
1154 : :
1155 [ - + ]: 5 : RESUME_INTERRUPTS();
1156 : 5 : }
1157 : :
1158 : : /*
1159 : : * Send the data to the specified parallel apply worker via shared-memory
1160 : : * queue.
1161 : : *
1162 : : * Returns false if the attempt to send data via shared memory times out, true
1163 : : * otherwise.
1164 : : */
1165 : : bool
1166 : 63920 : pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
1167 : : {
1168 : : int rc;
1169 : : shm_mq_result result;
1170 : 63920 : TimestampTz startTime = 0;
1171 : :
1172 [ - + ]: 63920 : Assert(!IsTransactionState());
1173 [ - + ]: 63920 : Assert(!winfo->serialize_changes);
1174 : :
1175 : : /*
1176 : : * We don't try to send data to parallel worker for 'immediate' mode. This
1177 : : * is primarily used for testing purposes.
1178 : : */
1094 peter@eisentraut.org 1179 [ + + ]: 63920 : if (unlikely(debug_logical_replication_streaming == DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE))
1302 akapila@postgresql.o 1180 : 4 : return false;
1181 : :
1182 : : /*
1183 : : * This timeout is a bit arbitrary but testing revealed that it is sufficient
1184 : : * to send the message unless the parallel apply worker is waiting on some
1185 : : * lock or there is a serious resource crunch. See the comments atop this file
1186 : : * to know why we are using a non-blocking way to send the message.
1187 : : */
1188 : : #define SHM_SEND_RETRY_INTERVAL_MS 1000
1189 : : #define SHM_SEND_TIMEOUT_MS (10000 - SHM_SEND_RETRY_INTERVAL_MS)
1190 : :
1191 : : for (;;)
1192 : : {
1326 1193 : 63916 : result = shm_mq_send(winfo->mq_handle, nbytes, data, true, true);
1194 : :
1195 [ + - ]: 63916 : if (result == SHM_MQ_SUCCESS)
1196 : 63916 : return true;
1326 akapila@postgresql.o 1197 [ # # ]:UBC 0 : else if (result == SHM_MQ_DETACHED)
1198 [ # # ]: 0 : ereport(ERROR,
1199 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1200 : : errmsg("could not send data to shared-memory queue")));
1201 : :
1202 [ # # ]: 0 : Assert(result == SHM_MQ_WOULD_BLOCK);
1203 : :
1204 : : /* Wait before retrying. */
1205 : 0 : rc = WaitLatch(MyLatch,
1206 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1207 : : SHM_SEND_RETRY_INTERVAL_MS,
1208 : : WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
1209 : :
1210 [ # # ]: 0 : if (rc & WL_LATCH_SET)
1211 : : {
1212 : 0 : ResetLatch(MyLatch);
1213 [ # # ]: 0 : CHECK_FOR_INTERRUPTS();
1214 : : }
1215 : :
1216 [ # # ]: 0 : if (startTime == 0)
1217 : 0 : startTime = GetCurrentTimestamp();
1218 [ # # ]: 0 : else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
1219 : : SHM_SEND_TIMEOUT_MS))
1220 : 0 : return false;
1221 : : }
1222 : : }
1223 : :
1224 : : /*
1225 : : * Switch to PARTIAL_SERIALIZE mode for the current transaction -- this means
1226 : : * that the current data and any subsequent data for this transaction will be
1227 : : * serialized to a file. This is done to prevent possible deadlocks with
1228 : : * another parallel apply worker (refer to the comments atop this file).
1229 : : */
1230 : : void
1326 akapila@postgresql.o 1231 :CBC 4 : pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
1232 : : bool stream_locked)
1233 : : {
1302 1234 [ + - ]: 4 : ereport(LOG,
1235 : : (errmsg("logical replication apply worker will serialize the remaining changes of remote transaction %u to a file",
1236 : : winfo->shared->xid)));
1237 : :
1238 : : /*
1239 : : * The parallel apply worker could be stuck for some reason (say waiting
1240 : : * on some lock by other backend), so stop trying to send data directly to
1241 : : * it and start serializing data to the file instead.
1242 : : */
1326 1243 : 4 : winfo->serialize_changes = true;
1244 : :
1245 : : /* Initialize the stream fileset. */
1246 : 4 : stream_start_internal(winfo->shared->xid, true);
1247 : :
1248 : : /*
1249 : : * Acquires the stream lock if not already to make sure that the parallel
1250 : : * apply worker will wait for the leader to release the stream lock until
1251 : : * the end of the transaction.
1252 : : */
1253 [ + - ]: 4 : if (!stream_locked)
1254 : 4 : pa_lock_stream(winfo->shared->xid, AccessExclusiveLock);
1255 : :
1256 : 4 : pa_set_fileset_state(winfo->shared, FS_SERIALIZE_IN_PROGRESS);
1257 : 4 : }
1258 : :
1259 : : /*
1260 : : * Wait until the parallel apply worker's transaction state has reached or
1261 : : * exceeded the given xact_state.
1262 : : */
1263 : : static void
1264 : 27 : pa_wait_for_xact_state(ParallelApplyWorkerInfo *winfo,
1265 : : ParallelTransState xact_state)
1266 : : {
1267 : : for (;;)
1268 : : {
1269 : : /*
1270 : : * Stop if the transaction state has reached or exceeded the given
1271 : : * xact_state.
1272 : : */
1273 [ + + ]: 197 : if (pa_get_xact_state(winfo->shared) >= xact_state)
1274 : 27 : break;
1275 : :
1276 : : /* Wait to be signalled. */
1277 : 170 : (void) WaitLatch(MyLatch,
1278 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1279 : : 10L,
1280 : : WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
1281 : :
1282 : : /* Reset the latch so we don't spin. */
1283 : 170 : ResetLatch(MyLatch);
1284 : :
1285 : : /* An interrupt may have occurred while we were waiting. */
1286 [ - + ]: 170 : CHECK_FOR_INTERRUPTS();
1287 : : }
1288 : 27 : }
1289 : :
1290 : : /*
1291 : : * Wait until the parallel apply worker's transaction finishes.
1292 : : */
1293 : : static void
1294 : 27 : pa_wait_for_xact_finish(ParallelApplyWorkerInfo *winfo)
1295 : : {
1296 : : /*
1297 : : * Wait until the parallel apply worker set the state to
1298 : : * PARALLEL_TRANS_STARTED which means it has acquired the transaction
1299 : : * lock. This is to prevent leader apply worker from acquiring the
1300 : : * transaction lock earlier than the parallel apply worker.
1301 : : */
1302 : 27 : pa_wait_for_xact_state(winfo, PARALLEL_TRANS_STARTED);
1303 : :
1304 : : /*
1305 : : * Wait for the transaction lock to be released. This is required to
1306 : : * detect deadlock among leader and parallel apply workers. Refer to the
1307 : : * comments atop this file.
1308 : : */
1309 : 27 : pa_lock_transaction(winfo->shared->xid, AccessShareLock);
1310 : 25 : pa_unlock_transaction(winfo->shared->xid, AccessShareLock);
1311 : :
1312 : : /*
1313 : : * Check if the state becomes PARALLEL_TRANS_FINISHED in case the parallel
1314 : : * apply worker failed while applying changes causing the lock to be
1315 : : * released.
1316 : : */
1317 [ - + ]: 25 : if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
1326 akapila@postgresql.o 1318 [ # # ]:UBC 0 : ereport(ERROR,
1319 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1320 : : errmsg("lost connection to the logical replication parallel apply worker")));
1326 akapila@postgresql.o 1321 :CBC 25 : }
1322 : :
1323 : : /*
1324 : : * Set the transaction state for a given parallel apply worker.
1325 : : */
1326 : : void
1327 : 54 : pa_set_xact_state(ParallelApplyWorkerShared *wshared,
1328 : : ParallelTransState xact_state)
1329 : : {
1330 : 54 : SpinLockAcquire(&wshared->mutex);
1331 : 54 : wshared->xact_state = xact_state;
1332 : 54 : SpinLockRelease(&wshared->mutex);
1333 : 54 : }
1334 : :
1335 : : /*
1336 : : * Get the transaction state for a given parallel apply worker.
1337 : : */
1338 : : static ParallelTransState
1339 : 247 : pa_get_xact_state(ParallelApplyWorkerShared *wshared)
1340 : : {
1341 : : ParallelTransState xact_state;
1342 : :
1343 : 247 : SpinLockAcquire(&wshared->mutex);
1344 : 247 : xact_state = wshared->xact_state;
1345 : 247 : SpinLockRelease(&wshared->mutex);
1346 : :
1347 : 247 : return xact_state;
1348 : : }
1349 : :
1350 : : /*
1351 : : * Cache the parallel apply worker information.
1352 : : */
1353 : : void
1354 : 518 : pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo)
1355 : : {
1356 : 518 : stream_apply_worker = winfo;
1357 : 518 : }
1358 : :
1359 : : /*
1360 : : * Form a unique savepoint name for the streaming transaction.
1361 : : *
1362 : : * Note that different subscriptions for publications on different nodes can
1363 : : * receive same remote xid, so we need to use subscription id along with it.
1364 : : *
1365 : : * Returns the name in the supplied buffer.
1366 : : */
1367 : : static void
1368 : 27 : pa_savepoint_name(Oid suboid, TransactionId xid, char *spname, Size szsp)
1369 : : {
1370 : 27 : snprintf(spname, szsp, "pg_sp_%u_%u", suboid, xid);
1371 : 27 : }
1372 : :
1373 : : /*
1374 : : * Define a savepoint for a subxact in parallel apply worker if needed.
1375 : : *
1376 : : * The parallel apply worker can figure out if a new subtransaction was
1377 : : * started by checking if the new change arrived with a different xid. In that
1378 : : * case define a named savepoint, so that we are able to rollback to it
1379 : : * if required.
1380 : : */
1381 : : void
1382 : 63403 : pa_start_subtrans(TransactionId current_xid, TransactionId top_xid)
1383 : : {
1384 [ + + ]: 63403 : if (current_xid != top_xid &&
1385 [ + + ]: 52 : !list_member_xid(subxactlist, current_xid))
1386 : : {
1387 : : MemoryContext oldctx;
1388 : : char spname[NAMEDATALEN];
1389 : :
1390 : 17 : pa_savepoint_name(MySubscription->oid, current_xid,
1391 : : spname, sizeof(spname));
1392 : :
1393 [ + + ]: 17 : elog(DEBUG1, "defining savepoint %s in logical replication parallel apply worker", spname);
1394 : :
1395 : : /* We must be in transaction block to define the SAVEPOINT. */
1396 [ + + ]: 17 : if (!IsTransactionBlock())
1397 : : {
1398 [ - + ]: 5 : if (!IsTransactionState())
1326 akapila@postgresql.o 1399 :UBC 0 : StartTransactionCommand();
1400 : :
1326 akapila@postgresql.o 1401 :CBC 5 : BeginTransactionBlock();
1402 : 5 : CommitTransactionCommand();
1403 : : }
1404 : :
1405 : 17 : DefineSavepoint(spname);
1406 : :
1407 : : /*
1408 : : * CommitTransactionCommand is needed to start a subtransaction after
1409 : : * issuing a SAVEPOINT inside a transaction block (see
1410 : : * StartSubTransaction()).
1411 : : */
1412 : 17 : CommitTransactionCommand();
1413 : :
1414 : 17 : oldctx = MemoryContextSwitchTo(TopTransactionContext);
1415 : 17 : subxactlist = lappend_xid(subxactlist, current_xid);
1416 : 17 : MemoryContextSwitchTo(oldctx);
1417 : : }
1418 : 63403 : }
1419 : :
1420 : : /* Reset the list that maintains subtransactions. */
1421 : : void
1422 : 25 : pa_reset_subtrans(void)
1423 : : {
1424 : : /*
1425 : : * We don't need to free this explicitly as the allocated memory will be
1426 : : * freed at the transaction end.
1427 : : */
1428 : 25 : subxactlist = NIL;
1429 : 25 : }
1430 : :
1431 : : /*
1432 : : * Handle STREAM ABORT message when the transaction was applied in a parallel
1433 : : * apply worker.
1434 : : */
1435 : : void
1436 : 12 : pa_stream_abort(LogicalRepStreamAbortData *abort_data)
1437 : : {
1438 : 12 : TransactionId xid = abort_data->xid;
1439 : 12 : TransactionId subxid = abort_data->subxid;
1440 : :
1441 : : /*
1442 : : * Update origin state so we can restart streaming from correct position
1443 : : * in case of crash.
1444 : : */
211 msawada@postgresql.o 1445 : 12 : replorigin_xact_state.origin_lsn = abort_data->abort_lsn;
1446 : 12 : replorigin_xact_state.origin_timestamp = abort_data->abort_time;
1447 : :
1448 : : /*
1449 : : * If the two XIDs are the same, it's in fact abort of toplevel xact, so
1450 : : * just free the subxactlist.
1451 : : */
1326 akapila@postgresql.o 1452 [ + + ]: 12 : if (subxid == xid)
1453 : : {
1454 : 2 : pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_FINISHED);
1455 : :
1456 : : /*
1457 : : * Release the lock as we might be processing an empty streaming
1458 : : * transaction in which case the lock won't be released during
1459 : : * transaction rollback.
1460 : : *
1461 : : * Note that it's ok to release the transaction lock before aborting
1462 : : * the transaction because even if the parallel apply worker dies due
1463 : : * to crash or some other reason, such a transaction would still be
1464 : : * considered aborted.
1465 : : */
1466 : 2 : pa_unlock_transaction(xid, AccessExclusiveLock);
1467 : :
1468 : 2 : AbortCurrentTransaction();
1469 : :
1470 [ + + ]: 2 : if (IsTransactionBlock())
1471 : : {
1472 : 1 : EndTransactionBlock(false);
1473 : 1 : CommitTransactionCommand();
1474 : : }
1475 : :
1476 : 2 : pa_reset_subtrans();
1477 : :
1478 : 2 : pgstat_report_activity(STATE_IDLE, NULL);
1479 : : }
1480 : : else
1481 : : {
1482 : : /* OK, so it's a subxact. Rollback to the savepoint. */
1483 : : int i;
1484 : : char spname[NAMEDATALEN];
1485 : :
1486 : 10 : pa_savepoint_name(MySubscription->oid, subxid, spname, sizeof(spname));
1487 : :
1488 [ + + ]: 10 : elog(DEBUG1, "rolling back to savepoint %s in logical replication parallel apply worker", spname);
1489 : :
1490 : : /*
1491 : : * Search the subxactlist, determine the offset tracked for the
1492 : : * subxact, and truncate the list.
1493 : : *
1494 : : * Note that for an empty sub-transaction we won't find the subxid
1495 : : * here.
1496 : : */
1497 [ + + ]: 12 : for (i = list_length(subxactlist) - 1; i >= 0; i--)
1498 : : {
1499 : 11 : TransactionId xid_tmp = lfirst_xid(list_nth_cell(subxactlist, i));
1500 : :
1501 [ + + ]: 11 : if (xid_tmp == subxid)
1502 : : {
1503 : 9 : RollbackToSavepoint(spname);
1504 : 9 : CommitTransactionCommand();
1505 : 9 : subxactlist = list_truncate(subxactlist, i);
1506 : 9 : break;
1507 : : }
1508 : : }
1509 : : }
1510 : 12 : }
1511 : :
1512 : : /*
1513 : : * Set the fileset state for a particular parallel apply worker. The fileset
1514 : : * will be set once the leader worker serialized all changes to the file
1515 : : * so that it can be used by parallel apply worker.
1516 : : */
1517 : : void
1518 : 16 : pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
1519 : : PartialFileSetState fileset_state)
1520 : : {
1521 : 16 : SpinLockAcquire(&wshared->mutex);
1522 : 16 : wshared->fileset_state = fileset_state;
1523 : :
1524 [ + + ]: 16 : if (fileset_state == FS_SERIALIZE_DONE)
1525 : : {
1526 [ - + ]: 4 : Assert(am_leader_apply_worker());
1527 [ - + ]: 4 : Assert(MyLogicalRepWorker->stream_fileset);
1528 : 4 : wshared->fileset = *MyLogicalRepWorker->stream_fileset;
1529 : : }
1530 : :
1531 : 16 : SpinLockRelease(&wshared->mutex);
1532 : 16 : }
1533 : :
1534 : : /*
1535 : : * Get the fileset state for the current parallel apply worker.
1536 : : */
1537 : : static PartialFileSetState
1538 : 87 : pa_get_fileset_state(void)
1539 : : {
1540 : : PartialFileSetState fileset_state;
1541 : :
1542 [ - + ]: 87 : Assert(am_parallel_apply_worker());
1543 : :
1544 : 87 : SpinLockAcquire(&MyParallelShared->mutex);
1545 : 87 : fileset_state = MyParallelShared->fileset_state;
1546 : 87 : SpinLockRelease(&MyParallelShared->mutex);
1547 : :
1548 : 87 : return fileset_state;
1549 : : }
1550 : :
1551 : : /*
1552 : : * Helper functions to acquire and release a lock for each stream block.
1553 : : *
1554 : : * Set locktag_field4 to PARALLEL_APPLY_LOCK_STREAM to indicate that it's a
1555 : : * stream lock.
1556 : : *
1557 : : * Refer to the comments atop this file to see how the stream lock is used.
1558 : : */
1559 : : void
1560 : 290 : pa_lock_stream(TransactionId xid, LOCKMODE lockmode)
1561 : : {
1562 : 290 : LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
1563 : : PARALLEL_APPLY_LOCK_STREAM, lockmode);
1564 : 288 : }
1565 : :
1566 : : void
1567 : 286 : pa_unlock_stream(TransactionId xid, LOCKMODE lockmode)
1568 : : {
1569 : 286 : UnlockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
1570 : : PARALLEL_APPLY_LOCK_STREAM, lockmode);
1571 : 286 : }
1572 : :
1573 : : /*
1574 : : * Helper functions to acquire and release a lock for each local transaction
1575 : : * apply.
1576 : : *
1577 : : * Set locktag_field4 to PARALLEL_APPLY_LOCK_XACT to indicate that it's a
1578 : : * transaction lock.
1579 : : *
1580 : : * Note that all the callers must pass a remote transaction ID instead of a
1581 : : * local transaction ID as xid. This is because the local transaction ID will
1582 : : * only be assigned while applying the first change in the parallel apply but
1583 : : * it's possible that the first change in the parallel apply worker is blocked
1584 : : * by a concurrently executing transaction in another parallel apply worker. We
1585 : : * can only communicate the local transaction id to the leader after applying
1586 : : * the first change so it won't be able to wait after sending the xact finish
1587 : : * command using this lock.
1588 : : *
1589 : : * Refer to the comments atop this file to see how the transaction lock is
1590 : : * used.
1591 : : */
1592 : : void
1593 : 56 : pa_lock_transaction(TransactionId xid, LOCKMODE lockmode)
1594 : : {
1595 : 56 : LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
1596 : : PARALLEL_APPLY_LOCK_XACT, lockmode);
1597 : 54 : }
1598 : :
1599 : : void
1600 : 50 : pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
1601 : : {
1602 : 50 : UnlockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
1603 : : PARALLEL_APPLY_LOCK_XACT, lockmode);
1604 : 50 : }
1605 : :
1606 : : /*
1607 : : * Decrement the number of pending streaming blocks and wait on the stream lock
1608 : : * if there is no pending block available.
1609 : : */
1610 : : void
1611 : 258 : pa_decr_and_wait_stream_block(void)
1612 : : {
1613 [ - + ]: 258 : Assert(am_parallel_apply_worker());
1614 : :
1615 : : /*
1616 : : * It is only possible to not have any pending stream chunks when we are
1617 : : * applying spooled messages.
1618 : : */
1619 [ + + ]: 258 : if (pg_atomic_read_u32(&MyParallelShared->pending_stream_count) == 0)
1620 : : {
1621 [ + - ]: 16 : if (pa_has_spooled_message_pending())
1622 : 16 : return;
1623 : :
1326 akapila@postgresql.o 1624 [ # # ]:UBC 0 : elog(ERROR, "invalid pending streaming chunk 0");
1625 : : }
1626 : :
1326 akapila@postgresql.o 1627 [ + + ]:CBC 242 : if (pg_atomic_sub_fetch_u32(&MyParallelShared->pending_stream_count, 1) == 0)
1628 : : {
1629 : 33 : pa_lock_stream(MyParallelShared->xid, AccessShareLock);
1630 : 31 : pa_unlock_stream(MyParallelShared->xid, AccessShareLock);
1631 : : }
1632 : : }
1633 : :
1634 : : /*
1635 : : * Finish processing the streaming transaction in the leader apply worker.
1636 : : */
1637 : : void
1638 : 27 : pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
1639 : : {
1640 [ - + ]: 27 : Assert(am_leader_apply_worker());
1641 : :
1642 : : /*
1643 : : * Unlock the shared object lock so that parallel apply worker can
1644 : : * continue to receive and apply changes.
1645 : : */
1646 : 27 : pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
1647 : :
1648 : : /*
1649 : : * Wait for that worker to finish. This is necessary to maintain commit
1650 : : * order which avoids failures due to transaction dependencies and
1651 : : * deadlocks.
1652 : : */
1653 : 27 : pa_wait_for_xact_finish(winfo);
1654 : :
294 alvherre@kurilemu.de 1655 [ + + ]: 25 : if (XLogRecPtrIsValid(remote_lsn))
1326 akapila@postgresql.o 1656 : 23 : store_flush_position(remote_lsn, winfo->shared->last_commit_end);
1657 : :
1658 : 25 : pa_free_worker(winfo);
1659 : 25 : }
|
