mirror of
https://github.com/postgres/postgres.git
synced 2026-01-27 21:43:08 +03:00
e385a4e2fd
Commit5b148706c5exposed functionality that allows multiple processes to use the same replication origin, enabling non-builtin logical replication solutions to implement parallel apply for large transactions. With this functionality, if two backends acquire the same replication origin and one of them resets it first, the acquired_by flag is cleared without acknowledging that another backend is still actively using the origin. This can lead to the origin being unintentionally dropped. If the shared memory for that dropped origin is later reused for a newly created origin, the remaining backend that still holds a pointer to the old memory may inadvertently advance the LSN of a completely different origin, causing unpredictable behavior. Although the underlying issue predates commit5b148706c5, it did not surface earlier because the internal parallel apply worker mechanism correctly coordinated origin resets and drops. This commit resolves the problem by introducing a reference counter for replication origins. The reference count increases when a backend sets the origin and decreases when it resets it. Additionally, the backend that first acquires the origin will not release it until all other backends using the origin have released it as well. The patch also prevents dropping a replication origin when acquired_by is zero but the reference counter is nonzero, covering the scenario where the first session exits without properly releasing the origin. Author: Hou Zhijie <houzj.fnst@fujitsu.com> Author: Hayato Kuroda <kuroda.hayato@fujitsu.com> Reviewed-by: Shveta Malik <shveta.malik@gmail.com> Reviewed-by: Amit Kapila <amit.kapila16@gmail.com> Discussion: https://postgr.es/m/TY4PR01MB169077EE72ABE9E55BAF162D494B5A@TY4PR01MB16907.jpnprd01.prod.outlook.com Discussion: https://postgr.es/m/CAMPB6wfe4zLjJL8jiZV5kjjpwBM2=rTRme0UCL7Ra4L8MTVdOg@mail.gmail.com
124 lines
3.5 KiB
Plaintext
124 lines
3.5 KiB
Plaintext
Parsed test spec with 2 sessions
|
|
|
|
starting permutation: s0_setup s0_is_setup s1_setup s1_is_setup s0_add_message s0_store_lsn s1_add_message s1_store_lsn s0_compare s1_reset s0_reset
|
|
step s0_setup: SELECT pg_replication_origin_session_setup('origin');
|
|
pg_replication_origin_session_setup
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
step s0_is_setup: SELECT pg_replication_origin_session_is_setup();
|
|
pg_replication_origin_session_is_setup
|
|
--------------------------------------
|
|
t
|
|
(1 row)
|
|
|
|
step s1_setup:
|
|
SELECT pg_replication_origin_session_setup('origin', pid)
|
|
FROM pg_stat_activity
|
|
WHERE application_name = 'isolation/parallel_session_origin/s0';
|
|
|
|
pg_replication_origin_session_setup
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
step s1_is_setup: SELECT pg_replication_origin_session_is_setup();
|
|
pg_replication_origin_session_is_setup
|
|
--------------------------------------
|
|
t
|
|
(1 row)
|
|
|
|
step s0_add_message:
|
|
SELECT 1
|
|
FROM pg_logical_emit_message(true, 'prefix', 'message on s0');
|
|
|
|
?column?
|
|
--------
|
|
1
|
|
(1 row)
|
|
|
|
step s0_store_lsn:
|
|
INSERT INTO local_lsn_store
|
|
SELECT 0, local_lsn FROM pg_replication_origin_status;
|
|
|
|
step s1_add_message:
|
|
SELECT 1
|
|
FROM pg_logical_emit_message(true, 'prefix', 'message on s1');
|
|
|
|
?column?
|
|
--------
|
|
1
|
|
(1 row)
|
|
|
|
step s1_store_lsn:
|
|
INSERT INTO local_lsn_store
|
|
SELECT 1, local_lsn FROM pg_replication_origin_status;
|
|
|
|
step s0_compare:
|
|
SELECT s0.lsn < s1.lsn
|
|
FROM local_lsn_store as s0, local_lsn_store as s1
|
|
WHERE s0.session = 0 AND s1.session = 1;
|
|
|
|
?column?
|
|
--------
|
|
t
|
|
(1 row)
|
|
|
|
step s1_reset: SELECT pg_replication_origin_session_reset();
|
|
pg_replication_origin_session_reset
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
step s0_reset: SELECT pg_replication_origin_session_reset();
|
|
pg_replication_origin_session_reset
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
|
|
starting permutation: s0_setup s0_is_setup s1_setup s1_is_setup s0_reset s1_reset s0_reset
|
|
step s0_setup: SELECT pg_replication_origin_session_setup('origin');
|
|
pg_replication_origin_session_setup
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
step s0_is_setup: SELECT pg_replication_origin_session_is_setup();
|
|
pg_replication_origin_session_is_setup
|
|
--------------------------------------
|
|
t
|
|
(1 row)
|
|
|
|
step s1_setup:
|
|
SELECT pg_replication_origin_session_setup('origin', pid)
|
|
FROM pg_stat_activity
|
|
WHERE application_name = 'isolation/parallel_session_origin/s0';
|
|
|
|
pg_replication_origin_session_setup
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
step s1_is_setup: SELECT pg_replication_origin_session_is_setup();
|
|
pg_replication_origin_session_is_setup
|
|
--------------------------------------
|
|
t
|
|
(1 row)
|
|
|
|
step s0_reset: SELECT pg_replication_origin_session_reset();
|
|
ERROR: cannot reset replication origin with ID 1 because it is still in use by other processes
|
|
step s1_reset: SELECT pg_replication_origin_session_reset();
|
|
pg_replication_origin_session_reset
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|
|
step s0_reset: SELECT pg_replication_origin_session_reset();
|
|
pg_replication_origin_session_reset
|
|
-----------------------------------
|
|
|
|
(1 row)
|
|
|