diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c index df3aad43e1..13513f2e57 100644 --- a/src/backend/replication/logical/applyparallelworker.c +++ b/src/backend/replication/logical/applyparallelworker.c @@ -7,43 +7,98 @@ * IDENTIFICATION * src/backend/replication/logical/applyparallelworker.c * - * This file contains routines that are intended to support setting up, using, - * and tearing down a ParallelApplyWorkerInfo. + * This file contains the code to launch, set up, and teardown parallel apply + * worker which receives the changes from the leader worker and invokes routines + * to apply those on the subscriber database. * - * Refer to the comments in the file header of logical/worker.c to see more - * information about parallel apply workers. + * This file contains routines that are intended to support setting up, using + * and tearing down a ParallelApplyWorkerInfo which is required to communicate + * among leader and parallel apply workers. + * + * The parallel apply workers are assigned (if available) as soon as xact's + * first stream is received for subscriptions that have set their 'streaming' + * option as parallel. The leader apply worker will send changes to this new + * worker via shared memory. We keep this worker assigned till the transaction + * commit is received and also wait for the worker to finish at commit. This + * preserves commit ordering and avoid file I/O in most cases, although we + * still need to spill to a file if there is no worker available. See comments + * atop logical/worker to know more about streamed xacts whose changes are + * spilled to disk. It is important to maintain commit order to avoid failures + * due to (a) transaction dependencies, say if we insert a row in the first + * transaction and update it in the second transaction on publisher then + * allowing the subscriber to apply both in parallel can lead to failure in the + * update. (b) deadlocks, allowing transactions that update the same set of + * rows/tables in the opposite order to be applied in parallel can lead to + * deadlocks. + * + * We maintain a worker pool to avoid restarting workers for each streaming + * transaction. We maintain each worker's information in the + * ParallelApplyWorkersList. After successfully launching a new worker, its + * information is added to the ParallelApplyWorkersList. Once the worker + * finishes applying the transaction, we mark it available for re-use. Now, + * before starting a new worker to apply the streaming transaction, we check + * the list for any available worker. Note that we maintain a maximum of half + * the max_parallel_apply_workers_per_subscription workers in the pool and + * after that, we simply exit the worker after applying the transaction. + * + * XXX This worker pool threshold is a bit arbitrary and we can provide a GUC + * variable for this in the future if required. + * + * The leader apply worker will create a separate dynamic shared memory segment + * when each parallel apply worker starts. The reason for this design is that + * we cannot count how many workers will be started. It may be possible to + * allocate enough shared memory in one segment based on the maximum number of + * parallel apply workers (max_parallel_apply_workers_per_subscription), but + * this would waste memory if no process is actually started. + * + * The dynamic shared memory segment will contain (a) a shm_mq that is used to + * send changes in the transaction from leader apply worker to parallel apply + * worker (b) another shm_mq that is used to send errors (and other messages + * reported via elog/ereport) from the parallel apply worker to leader apply + * worker (c) necessary information to be shared among parallel apply workers + * and leader apply worker (i.e. members of ParallelApplyWorkerShared). * * Locking Considerations - * ------------------ - * Since the database structure (schema of subscription tables, etc.) of - * publisher and subscriber may be different, applying transactions in parallel - * mode on the subscriber side may cause some deadlock problems that do not - * occur on the publisher side. + * ---------------------- + * Since the database structure (schema of subscription tables, constraints, + * etc.) of the publisher and subscriber could be different, applying + * transactions in parallel mode on the subscriber side can cause some + * deadlocks that do not occur on the publisher side which is expected and can + * happen even without parallel mode. In order to detect the deadlocks among + * leader and parallel apply workers, we need to ensure that we wait using lmgr + * locks, otherwise, such deadlocks won't be detected. The other approach was + * to not allow parallelism when the schema of tables is different between the + * publisher and subscriber but that would be too restrictive and would require + * the publisher to send much more information than it is currently sending. * - * Suppose a subscribed table does not have a unique key on the publisher and - * has a unique key on the subscriber. + * Consider a case where the subscribed table does not have a unique key on the + * publisher and has a unique key on the subscriber. * * 1) Deadlock between the leader apply worker and a parallel apply worker * * The parallel apply worker (PA) is executing TX-1 and the leader apply worker * (LA) is executing TX-2 concurrently on the subscriber. Now, LA is waiting * for PA because of the unique key of the subscribed table while PA is waiting - * for LA to send further messages. + * for LA to send the next stream of changes or transaction finish command + * message. * * In order for lmgr to detect this, we have LA acquire a session lock on the * remote transaction (by pa_lock_stream()) and have PA wait on the lock before - * trying to receive messages. In other words, LA acquires the lock before - * sending STREAM_STOP and releases it if already acquired before sending - * STREAM_START, STREAM_ABORT (for toplevel transaction), STREAM_PREPARE and - * STREAM_COMMIT. For PA, it always needs to acquire the lock after processing - * STREAM_STOP and STREAM_ABORT (for subtransaction) and then release - * immediately after acquiring it. That way, when PA is waiting for LA, we can - * have a wait-edge from PA to LA in lmgr, which will make a deadlock in lmgr - * like: + * trying to receive the next stream of changes. Specifically, LA will acquire + * the lock before sending the STREAM_STOP and will release it if already + * acquired before sending the STREAM_START, STREAM_ABORT (for toplevel + * transaction), STREAM_PREPARE, and STREAM_COMMIT. The PA will acquire the + * lock after processing STREAM_STOP and STREAM_ABORT (for subtransaction) and + * then release the lock immediately after acquiring it. * + * The lock graph for the above example will look as follows: * LA (waiting to acquire the lock on the unique index) -> PA (waiting to * acquire the lock on the remote transaction) -> LA * + * This way, when PA is waiting for LA for the next stream of changes, we can + * have a wait-edge from PA to LA in lmgr, which will make us detect the + * deadlock between LA and PA. + * * 2) Deadlock between the leader apply worker and parallel apply workers * * This scenario is similar to the first case but TX-1 and TX-2 are executed by @@ -51,38 +106,46 @@ * PA-2 is waiting for PA-1 to complete its transaction while PA-1 is waiting * for subsequent input from LA. Also, LA is waiting for PA-2 to complete its * transaction in order to preserve the commit order. There is a deadlock among - * three processes. + * the three processes. + * + * In order for lmgr to detect this, we have PA acquire a session lock (this is + * a different lock than referred in previous case, see + * pa_lock_transaction()) on the transaction being applied and have LA wait on + * the lock before proceeding in the transaction finish commands. Specifically, + * PA will acquire this lock in Share mode before executing the first message + * of the transaction and release it at the xact end. LA will acquire this lock + * in AccessExclusive mode at transaction finish commands (STREAM_COMMIT and + * STREAM_PREAPRE) and release it immediately. * - * In order for lmgr to detect this, we have PA acquire a session lock on the - * transaction being applied and have LA wait on the lock before proceeding in - * transaction finish commands. That way, we can have a wait-edge from LA to PA - * in lmgr, which will make a deadlock in lmgr like: + * The lock graph for the above example will look as follows: + * LA (waiting to acquire the transaction lock) -> PA-2 (waiting to acquire the + * lock due to unique index constraint) -> PA-1 (waiting to acquire the stream + * lock) -> LA * - * LA (waiting to acquire the local transaction lock) -> PA-2 (waiting to - * acquire the lock on the unique index) -> PA-1 (waiting to acquire the lock - * on the remote transaction) -> LA + * This way when LA is waiting to finish the transaction end command to preserve + * the commit order, we will be able to detect deadlock, if any. * * One might think we can use XactLockTableWait(), but XactLockTableWait() * considers PREPARED TRANSACTION as still in progress which means the lock - * won't be released even if the parallel apply worker prepared the transaction. + * won't be released even after the parallel apply worker has prepared the + * transaction. * * 3) Deadlock when the shm_mq buffer is full * - * The scenario where LA has to wait is when the shm_mq buffer is full. In the - * above scenario (ie. PA-1 and PA-2 are executing transactions concurrently), - * if the shm_mq buffer between LA and PA-2 is full, LA has to wait to send - * messages, and this wait doesn't appear in lmgr. + * In the previous scenario (ie. PA-1 and PA-2 are executing transactions + * concurrently), if the shm_mq buffer between LA and PA-2 is full, LA has to + * wait to send messages, and this wait doesn't appear in lmgr. * * To resolve this issue, we use non-blocking write and wait with a timeout. If - * timeout is exceeded, the LA reports an error and restarts logical + * the timeout is exceeded, the LA reports an error and restarts logical * replication. * - * 4) Lock type + * 4) Lock types * * Both the stream lock and the transaction lock mentioned above are - * session-level locks, because both locks could be acquired outside the + * session-level locks because both locks could be acquired outside the * transaction, and the stream lock in the leader need to persist across - * transaction boundaries until the end of the streaming transaction. + * transaction boundaries i.e until the end of the streaming transaction. *------------------------------------------------------------------------- */ @@ -158,7 +221,7 @@ static HTAB *ParallelApplyWorkersHash = NULL; * list entry is removed if there are already enough workers in the worker * pool either at the end of the transaction or while trying to find a free * worker for applying the transaction. For more information about the worker - * pool, see comments atop worker.c. + * pool, see comments atop this file. */ static List *ParallelApplyWorkersList = NIL; @@ -365,7 +428,7 @@ pa_find_worker(TransactionId xid) * and frees the corresponding info. Otherwise it just marks the worker as * available for reuse. * - * For more information about the worker pool, see comments atop worker.c. + * For more information about the worker pool, see comments atop this file. * * Returns true if the worker is stopped, false otherwise. */ @@ -726,9 +789,11 @@ ParallelApplyWorkerMain(Datum main_arg) LogicalParallelApplyLoop(mqh); /* - * The parallel apply worker should not get here, because the parallel - * apply worker should only stop when it receives a SIGTERM or SIGINT from - * the leader or detects a change in subscription information. + * The parallel apply worker must not get here because the parallel + * apply worker will only stop when it receives a SIGTERM or SIGINT from + * the leader, detects a change in subscription information, or when there + * there is an error. None of these cases will allow the code to reach + * here. */ Assert(false); } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 56e3c6c1a2..d9ef50864e 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -24,52 +24,10 @@ * Streamed transactions (large transactions exceeding a memory limit on the * upstream) are applied using one of two approaches: * - * 1) Parallel apply workers + * 1) Write to temporary files and apply when the final commit arrives * - * If streaming = parallel, we assign a new parallel apply worker (if - * available) as soon as the xact's first stream is received. The leader apply - * worker will send changes to this new worker via shared memory. We keep this - * worker assigned till the transaction commit is received and also wait for - * the worker to finish at commit. This preserves commit ordering and avoids - * file I/O in most cases, although we still need to spill to a file if there - * is no worker available. It is important to maintain commit order to avoid - * failures due to (a) transaction dependencies, say if we insert a row in the - * first transaction and update it in the second transaction then allowing to - * apply both in parallel can lead to failure in the update. (b) deadlocks, - * allowing transactions that update the same set of rows/tables in opposite - * order to be applied in parallel can lead to deadlocks. - * - * We maintain a worker pool to avoid restarting workers for each streaming - * transaction. We maintain each worker's information in the - * ParallelApplyWorkersList. After successfully launching a new worker, its - * information is added to the ParallelApplyWorkersList. Once the worker - * finishes applying the transaction, we mark it available for re-use. Now, - * before starting a new worker to apply the streaming transaction, we check - * the list for any available worker. Note that we maintain a maximum of half - * the max_parallel_apply_workers_per_subscription workers in the pool and - * after that, we simply exit the worker after applying the transaction. - * - * XXX This worker pool threshold is a bit arbitrary and we can provide a GUC - * variable for this in the future if required. - * - * The leader apply worker will create separate dynamic shared memory segment - * when each parallel apply worker starts. The reason for this design is that - * we cannot count how many workers will be started. It may be possible to - * allocate enough shared memory in one segment based on the maximum number of - * parallel apply workers (max_parallel_apply_workers_per_subscription), but this - * would waste memory if no process is actually started. - * - * The dynamic shared memory segment will contain (a) a shm_mq that is used to - * send changes in the transaction from leader apply worker to parallel apply - * worker (b) another shm_mq that is used to send errors (and other messages - * reported via elog/ereport) from the parallel apply worker to leader apply - * worker (c) necessary information to be shared among parallel apply workers - * and leader apply worker (i.e. members of ParallelApplyWorkerShared). - * - * If no parallel apply worker is available to handle the streamed transaction - * we follow approach 2. - * - * 2) Write to temporary files and apply when the final commit arrives + * This approach is used when user has set subscription's streaming option as + * on. * * Unlike the regular (non-streamed) case, handling streamed transactions has * to handle aborts of both the toplevel transaction and subtransactions. This @@ -96,6 +54,12 @@ * the file we desired across multiple stream-open calls for the same * transaction. * + * 2) Parallel apply workers. + * + * This approach is used when user has set subscription's streaming option as + * parallel. See logical/applyparallelworker.c for information about this + * approach. + * * TWO_PHASE TRANSACTIONS * ---------------------- * Two phase transactions are replayed at prepare and then committed or