Hi Alexander, Thanks for the script, it reproduced here as is.
On Fri, Oct 23, 2026 at 1:00 PM Alexander Lakhin <[email protected]> wrote: > That is, with v3 applied, a backend still gets stuck. It does, and so does master: your script gives the same result on master, 18.6 and 17.11, 3 runs out of 3 on each. DROP DATABASE waits for the backend in SyncRepWaitForLSN(), which holds interrupts and never absorbs the barrier. v1 terminated such a session, which shihao pointed out; since v2 it is left alone, because its transaction has already committed locally. The login blocks behind the two VACUUM FULLs queued on DROP DATABASE's lock on pg_subscription. The attached v5 is split in two. 0001 fixes your case on its own and has no new GUC: it absorbs the barrier inside the sync-rep wait (and in a paused COPY FROM STDIN), but only for barrier types that do nothing commit does not already do after its commit record. The commit message has the details, including io_method=io_uring. 0002 is the timeout part from before, on top of it. With 0001 alone and your script, DROP DATABASE completes in 41-83 ms and the test login in under 10 ms, while the waiter keeps waiting for the standby; once synchronous_standby_names is reset, its COMMIT returns without a WARNING. With a paused COPY instead, DROP completes in under 65 ms and the COPY goes on once the client sends the rest. Both cases are in 0001's own test, 061. In the coverage matrix, 0001 alone clears only the sync-rep and COPY controls, in 8-9 ms; the ClientWrite rows stay stuck as on master. With 0002 on top they clear after the timeout as with v3, and no control session is terminated. Regards, Manu
From ede9dfa4a47dfed3ad3e20d941f64d91125f6fb3 Mon Sep 17 00:00:00 2001 From: Manu <[email protected]> Date: Fri, 23 Oct 2026 15:24:21 -0300 Subject: [PATCH v5 1/2] Absorb ProcSignalBarriers while waiting for sync rep or client input WaitForProcSignalBarrier() waits until every process has absorbed the barrier, but two kinds of wait never absorb one. A backend waiting in SyncRepWaitForLSN() holds off interrupts, since its commit record has already been flushed, so CHECK_FOR_INTERRUPTS() does nothing there. A backend waiting for more data from its client in the middle of a command, as in COPY FROM STDIN, only services ProcDiePending in ProcessClientReadInterrupt(). Whatever waits for the barrier, such as DROP DATABASE, then waits for as long as they do. With VACUUM FULL of pg_subscription and pg_authid queued behind DROP DATABASE's lock on pg_subscription, even new connections block, since rebuilding the relcache init files needs a lock on every catalog with a syscache. Add ProcessProcSignalBarrierNoError(), which absorbs the pending barriers only if every pending type can be absorbed where an ERROR is not allowed, and call it in both waits. Each barrier type now declares this in BarrierTypeAbsorbableWithoutError(), whose switch has no default case, so that a new type has to be classified. A type qualifies only if its processing does nothing that CommitTransaction() does not already do after the commit record. PROCSIGNAL_BARRIER_SMGRRELEASE closes files; with io_method=io_uring that may wait for in-flight IO and can raise an ERROR on an internal consistency failure, but smgrDoPendingDeletes() runs the same path after the commit record. PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO only sets a flag or reads shared state. The data checksum types do not qualify, because AbsorbDataChecksumsBarrier() raises an ERROR on an unexpected state, which after the commit record would be a PANIC; if one of them is pending, nothing is absorbed, as before. The sync rep waiter keeps waiting: its commit is acknowledged only once the standby confirms it. Reported-by: Alexander Lakhin <[email protected]> Discussion: https://postgr.es/m/caa4ek1lc09tq3vtr04ppxrbyqnmb66_pgyues2c_5zowfyr...@mail.gmail.com --- src/backend/replication/syncrep.c | 10 ++ src/backend/storage/ipc/procsignal.c | 108 ++++++++++++ src/backend/tcop/postgres.c | 11 ++ src/include/storage/procsignal.h | 6 + src/test/recovery/meson.build | 1 + .../t/061_procsignal_barrier_absorb.pl | 164 ++++++++++++++++++ 6 files changed, 300 insertions(+) create mode 100644 src/test/recovery/t/061_procsignal_barrier_absorb.pl diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index d870f09..572d747 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -82,6 +82,7 @@ #include "replication/walsender.h" #include "replication/walsender_private.h" #include "storage/proc.h" +#include "storage/procsignal.h" #include "tcop/tcopprot.h" #include "utils/guc_hooks.h" #include "utils/ps_status.h" @@ -334,6 +335,15 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) break; } + /* + * Interrupts are held off, so CHECK_FOR_INTERRUPTS() would not absorb + * a ProcSignalBarrier, and whoever waits for one would wait as long + * as we do. Absorb it here if every pending type allows it where an + * ERROR is not allowed: the transaction has already committed. + */ + if (ProcSignalBarrierPending) + ProcessProcSignalBarrierNoError(); + /* * Wait on latch. Any condition that should wake us up will set the * latch, so no need for timeout. diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index eb0f4d6..95a3269 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -124,6 +124,8 @@ static ProcSignalSlot *MyProcSignalSlot = NULL; static bool CheckProcSignal(ProcSignalReason reason); static void CleanupProcSignalState(int status, Datum arg); static void ResetProcSignalBarrierBits(uint32 flags); +static void ProcessProcSignalBarrierInternal(bool noerror); +static bool BarrierTypeAbsorbableWithoutError(ProcSignalBarrierType type); /* * ProcSignalShmemRequest @@ -511,6 +513,90 @@ HandleProcSignalBarrierInterrupt(void) */ void ProcessProcSignalBarrier(void) +{ + ProcessProcSignalBarrierInternal(false); +} + +/* + * ProcessProcSignalBarrierNoError - absorb barriers where an ERROR is not + * allowed + * + * Some waits cannot run CHECK_FOR_INTERRUPTS(): SyncRepWaitForLSN() holds off + * interrupts after the commit record has been flushed, and a backend waiting + * for more data from its client in the middle of a command, as in COPY FROM + * STDIN, cannot process a query cancel there. If such a process did not + * absorb barriers while it waits, every process waiting for a barrier, such as + * DROP DATABASE, would wait as long as it does. + * + * This absorbs the pending barriers only if every pending type can be absorbed + * where an ERROR is not allowed (see BarrierTypeAbsorbableWithoutError()). + * Otherwise it absorbs nothing and leaves all of them pending for the next + * CHECK_FOR_INTERRUPTS(), exactly as if it had not been called: the barrier + * generation can only advance once every pending type has been handled. + * + * It may be called with interrupts held off, but not in a critical section, + * and not while holding an LWLock. + */ +void +ProcessProcSignalBarrierNoError(void) +{ + Assert(CritSectionCount == 0); + + ProcessProcSignalBarrierInternal(true); +} + +/* + * Can this barrier type be absorbed where an ERROR is not allowed? + * + * ProcessProcSignalBarrierNoError() absorbs the pending barriers only if this + * returns true for every pending type. Return true only for a type whose + * processing does nothing that CommitTransaction() does not already do after + * its commit record: it must not raise an ERROR, allocate memory or take a lock + * that the caller could be holding, except on a path that commit processing + * itself runs in the same state. There is deliberately no default case, so + * that the compiler warns if a new type is not classified here. + */ +static bool +BarrierTypeAbsorbableWithoutError(ProcSignalBarrierType type) +{ + switch (type) + { + case PROCSIGNAL_BARRIER_SMGRRELEASE: + + /* + * smgrreleaseall() closes files without allocating memory (see + * _fdvec_resize()). With io_method=io_uring, FileClose() may + * wait for this backend's in-flight IO in pgaio_closing_fd(), + * which can raise an ERROR only on an internal consistency + * failure. CommitTransaction() runs the same path after its + * commit record, in smgrDoPendingDeletes(). + */ + return true; + case PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO: + + /* + * With a transaction ID assigned this only sets a flag, consumed + * by AtEOXact_LogicalCtl(); otherwise it reads shared state under + * a shared LWLock that no barrier emitter holds while it waits. + */ + return true; + case PROCSIGNAL_BARRIER_CHECKSUM_OFF: + case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON: + case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF: + case PROCSIGNAL_BARRIER_CHECKSUM_ON: + /* AbsorbDataChecksumsBarrier() raises an ERROR on a bad state. */ + return false; + } + + return false; /* not reached for a valid type */ +} + +/* + * Workhorse for ProcessProcSignalBarrier() and + * ProcessProcSignalBarrierNoError(). + */ +static void +ProcessProcSignalBarrierInternal(bool noerror) { uint64 local_gen; uint64 shared_gen; @@ -556,6 +642,28 @@ ProcessProcSignalBarrier(void) */ flags = pg_atomic_exchange_u32(&MyProcSignalSlot->pss_barrierCheckMask, 0); + /* + * Where an ERROR is not allowed, absorb nothing unless every pending type + * can be absorbed there: put all the bits back for the next call. + */ + if (noerror) + { + uint32 check = flags; + + while (check != 0) + { + ProcSignalBarrierType type; + + type = (ProcSignalBarrierType) pg_rightmost_one_pos32(check); + if (!BarrierTypeAbsorbableWithoutError(type)) + { + ResetProcSignalBarrierBits(flags); + return; + } + BARRIER_CLEAR_BIT(check, type); + } + } + /* * If there are no flags set, then we can skip doing any real work. * Otherwise, establish a PG_TRY block, so that we don't lose track of diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index b6bdfe2..b3ad8e4 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -544,6 +544,17 @@ ProcessClientReadInterrupt(bool blocked) else SetLatch(MyLatch); } + else if (blocked && ProcSignalBarrierPending) + { + /* + * Waiting for more data from the client in the middle of a command, + * as in COPY FROM STDIN. CHECK_FOR_INTERRUPTS() is not safe here, + * but barriers that can be absorbed where an ERROR is not allowed + * are, so that a client that stops sending does not hold up whoever + * waits for the barrier. + */ + ProcessProcSignalBarrierNoError(); + } errno = save_errno; } diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index aaa158b..7721814 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -45,6 +45,11 @@ typedef enum #define NUM_PROCSIGNALS (PROCSIG_RECOVERY_CONFLICT + 1) +/* + * Barrier types. A new type must also be classified in + * BarrierTypeAbsorbableWithoutError() (procsignal.c), which decides whether it + * can be absorbed where an ERROR is not allowed. + */ typedef enum { PROCSIGNAL_BARRIER_SMGRRELEASE, /* ask smgr to close files */ @@ -77,6 +82,7 @@ extern void SendCancelRequest(int backendPID, const uint8 *cancel_key, int cance extern uint64 EmitProcSignalBarrier(ProcSignalBarrierType type); extern void WaitForProcSignalBarrier(uint64 generation); extern void ProcessProcSignalBarrier(void); +extern void ProcessProcSignalBarrierNoError(void); extern void procsignal_sigusr1_handler(SIGNAL_ARGS); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index b6ac9dd..5732992 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -69,6 +69,7 @@ tests += { 't/058_shutdown_crash_restart.pl', 't/059_remote_apply_status_interval.pl', 't/060_multixact_crashed_updater.pl', + 't/061_procsignal_barrier_absorb.pl', ], }, } diff --git a/src/test/recovery/t/061_procsignal_barrier_absorb.pl b/src/test/recovery/t/061_procsignal_barrier_absorb.pl new file mode 100644 index 0000000..478a133 --- /dev/null +++ b/src/test/recovery/t/061_procsignal_barrier_absorb.pl @@ -0,0 +1,164 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Sessions that wait where they cannot run CHECK_FOR_INTERRUPTS() still absorb +# the ProcSignalBarriers that can be absorbed where an ERROR is not allowed (see +# BarrierTypeAbsorbableWithoutError()): a session waiting +# for synchronous replication after its commit, and one in COPY FROM STDIN +# waiting for more data from its client. DROP DATABASE must not wait for them, +# and they must go on waiting. A pending barrier whose processing can raise +# an ERROR, such as a data checksums change, is not absorbed there. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $node = PostgreSQL::Test::Cluster->new('primary'); +$node->init(allows_streaming => 1); +# A synchronous standby that never connects. It is set before the server +# starts, so that it is in effect before the first commit; only the session +# that waits for synchronous replication requests it. +$node->append_conf( + 'postgresql.conf', qq( +synchronous_standby_names = 'nosuch' +synchronous_commit = local +)); +$node->start; + +$node->safe_psql('postgres', 'CREATE TABLE copy_target (a int)'); +$node->safe_psql('postgres', 'CREATE DATABASE target1'); +$node->safe_psql('postgres', 'CREATE DATABASE target2'); + +# Start a psql running $sql in the background, with application_name $appname. +sub start_psql +{ + my ($appname, $sql, $in, $out, $err) = @_; + + return IPC::Run::start( + [ + 'psql', '--no-psqlrc', '--quiet', + '--dbname' => $node->connstr('postgres') + . " application_name=$appname", + defined $sql ? ('--command' => $sql) : () + ], + '<' => $in, + '>' => $out, + '2>' => $err); +} + +# Return the PID of the session $appname once its wait event is $wait_event. +sub session_pid +{ + my ($appname, $wait_event) = @_; + + $node->poll_query_until('postgres', + "SELECT wait_event = '$wait_event' FROM pg_stat_activity WHERE application_name = '$appname'" + ) or die "timed out waiting for session \"$appname\""; + return $node->safe_psql('postgres', + "SELECT pid FROM pg_stat_activity WHERE application_name = '$appname'"); +} + +# DROP DATABASE $dbname must complete while session $pid keeps waiting. +sub check_drop_does_not_wait +{ + my ($pid, $dbname, $what, $wait_event) = @_; + my $offset = -s $node->logfile; + my $timed_out = 0; + + # With a timeout, so that a regression fails here instead of hanging. + $node->psql( + 'postgres', + "DROP DATABASE $dbname", + timeout => $PostgreSQL::Test::Utils::timeout_default, + timed_out => \$timed_out); + ok(!$timed_out, "$what: DROP DATABASE did not wait for the session"); + is( $node->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = '$dbname'"), + '0', + "$what: DROP DATABASE completed"); + is( $node->safe_psql( + 'postgres', "SELECT wait_event FROM pg_stat_activity WHERE pid = $pid"), + $wait_event, + "$what: session is still waiting"); + ok( !$node->log_contains( + "still waiting for backend with PID $pid", $offset), + "$what: no wait for the barrier was reported"); +} + +# +# A session waiting for a synchronous standby that is not connected, after its +# transaction has committed locally. +# +my $log_offset = -s $node->logfile; +my ($sr_in, $sr_out, $sr_err) = ('', '', ''); +my $syncrep = start_psql('syncrep_waiter', + 'SET synchronous_commit = on; CREATE TABLE syncrep_target ()', + \$sr_in, \$sr_out, \$sr_err); +my $syncrep_pid = session_pid('syncrep_waiter', 'SyncRep'); +check_drop_does_not_wait($syncrep_pid, 'target1', 'synchronous replication', + 'SyncRep'); + +# Disabling data checksums emits barriers whose processing can raise an ERROR, +# so the waiting session must not absorb them: the operation waits for it. +if ($node->safe_psql('postgres', 'SHOW data_checksums') ne 'on') +{ + $node->safe_psql('postgres', 'SELECT pg_enable_data_checksums()'); + $node->poll_query_until('postgres', 'SHOW data_checksums', 'on') + or die 'timed out waiting for data checksums to be enabled'; +} +my $offset = -s $node->logfile; +$node->safe_psql('postgres', 'SELECT pg_disable_data_checksums()'); +ok( eval { + $node->wait_for_log( + qr/still waiting for backend with PID $syncrep_pid to accept ProcSignalBarrier/, + $offset); + 1; + }, + 'data checksums: the operation waits for the session'); +isnt($node->safe_psql('postgres', 'SHOW data_checksums'), + 'off', 'data checksums: not disabled while the session waits'); +is( $node->safe_psql( + 'postgres', + "SELECT wait_event FROM pg_stat_activity WHERE pid = $syncrep_pid"), + 'SyncRep', + 'data checksums: session is still waiting'); + +# Release the session: its commit is acknowledged, without a cancel, and the +# data checksums change goes through. +$node->safe_psql('postgres', + "ALTER SYSTEM SET synchronous_standby_names = ''"); +$node->reload; +$syncrep->finish; +ok( !$node->log_contains( + 'canceling the wait for synchronous replication', $log_offset), + 'synchronous replication: the wait was not canceled'); +is( $node->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_class WHERE relname = 'syncrep_target'"), + '1', + 'synchronous replication: transaction completed'); +ok($node->poll_query_until('postgres', 'SHOW data_checksums', 'off'), + 'data checksums: disabled once the session moved on'); + +# +# A COPY FROM STDIN whose client sent one row and then paused. +# +my ($copy_in, $copy_out, $copy_err) = ("COPY copy_target FROM STDIN;\n1\n", '', ''); +my $copy = start_psql('copy_paused', undef, \$copy_in, \$copy_out, \$copy_err); +$copy->pump_nb; +$node->poll_query_until('postgres', + "SELECT wait_event = 'ClientRead' AND state = 'active' FROM pg_stat_activity WHERE application_name = 'copy_paused'" +) or die 'timed out waiting for COPY FROM STDIN to start'; +my $copy_pid = session_pid('copy_paused', 'ClientRead'); +check_drop_does_not_wait($copy_pid, 'target2', 'COPY FROM STDIN', + 'ClientRead'); +$copy_in .= "2\n\\.\n"; +$copy->finish; +is($node->safe_psql('postgres', 'SELECT count(*) FROM copy_target'), + '2', 'COPY FROM STDIN: completed'); + +$node->stop; +done_testing(); -- 2.56.0
From 20c3d3de8a89619c155e32e7e6ba4457daa36383 Mon Sep 17 00:00:00 2001 From: Manu <[email protected]> Date: Fri, 23 Oct 2026 01:06:20 -0300 Subject: [PATCH v5 2/2] Terminate sessions blocked in ClientWrite that hold up a barrier WaitForProcSignalBarrier() waits until every process has absorbed the barrier. A client backend or WAL sender that is blocked sending data to a client that has stopped reading never does: while blocked in a client write it services only ProcDiePending (see ProcessClientWriteInterrupt()), not the barrier and not a recovery conflict signal. Whatever waits on the barrier then waits for as long as that client stays stuck. That includes the end of recovery and the replay of the record that enables logical decoding when wal_level=replica, so a standby can neither promote nor keep replaying, as well as changing data checksums online, and DROP DATABASE and the other users of PROCSIGNAL_BARRIER_SMGRRELEASE. Add procsignal_barrier_timeout (default 30s, 0 disables). A client backend or WAL sender that has not absorbed the barrier, is blocked writing to its client and has not completed a send to it for that long is sent SIGTERM, as pg_terminate_backend() would do, and the wait goes on until it exits. To tell a client that reads slowly from one that has stopped reading, each process counts its completed sends in a new PGPROC field, clientSendCount, which the waiter samples; a session sending a single large value to a slow client thus keeps going. So that a client reading just enough cannot hold the barrier forever, such a session is terminated anyway once the wait has lasted ten times the timeout. A session waiting for anything else is left alone. A backend waiting for synchronous replication or in COPY FROM STDIN absorbs the barrier while it waits since the previous commit, and other kinds of process are never signaled. A backend that blocks while it sends an error message holds off interrupts, so SIGTERM does not release it until the write completes or fails, just as without this patch. The waiter logs that termination was requested, and logs again if the process has not exited five seconds later, so as not to suggest that it is gone. The slot's PID and barrier generation are read together under its spinlock before signaling, so that a process that took over the slot after the one that did not absorb the barrier exited is not signaled. This applies to every waiter, not only to the startup process, since the hang is the same on a primary. Reported-by: shihao zhong <[email protected]> Suggested-by: Masahiko Sawada <[email protected]> Suggested-by: shihao zhong <[email protected]> Reviewed-by: shihao zhong <[email protected]> Discussion: https://postgr.es/m/caa4ek1lc09tq3vtr04ppxrbyqnmb66_pgyues2c_5zowfyr...@mail.gmail.com --- doc/src/sgml/config.sgml | 46 ++ src/backend/libpq/pqcomm.c | 6 + src/backend/storage/ipc/procsignal.c | 226 +++++++- src/backend/storage/lmgr/proc.c | 1 + src/backend/utils/misc/guc_parameters.dat | 10 + src/backend/utils/misc/guc_tables.c | 1 + src/backend/utils/misc/postgresql.conf.sample | 3 + src/include/storage/proc.h | 7 + src/include/storage/procsignal.h | 3 + src/test/recovery/meson.build | 1 + .../t/062_procsignal_barrier_timeout.pl | 526 ++++++++++++++++++ 11 files changed, 825 insertions(+), 5 deletions(-) create mode 100644 src/test/recovery/t/062_procsignal_barrier_timeout.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index f7db299..88f54e2 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11836,6 +11836,52 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-procsignal-barrier-timeout" xreflabel="procsignal_barrier_timeout"> + <term><varname>procsignal_barrier_timeout</varname> (<type>integer</type>) + <indexterm> + <primary><varname>procsignal_barrier_timeout</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Some operations change state that every running process must take + into account before the operation can continue, and wait until all + processes have done so. Among them are the end of recovery and the + replay of WAL on a standby when logical decoding is being enabled, + enabling or disabling data checksums online (see + <xref linkend="functions-admin-checksum"/>) and + <command>DROP DATABASE</command>. A session can only do this + between other work; one that is blocked sending data to a client that + has stopped reading cannot, and the operation would wait for it + indefinitely. This parameter sets how long a client backend or WAL + sender process that is blocked that way may go without sending + anything to its client before the operation terminates it, as + <function>pg_terminate_backend</function> would. A session that is + still sending, for example a single large value to a slow client, is + terminated only once the operation has waited ten times this long. + A session that is blocked while sending an error message to its client + does not exit when asked to; the operation keeps waiting for it until + the client reads or the connection is closed. Sessions waiting for + synchronous replication, or for more data from the client during + <command>COPY FROM STDIN</command>, take the change into account + while they wait, unless data checksums are being enabled or disabled + at the same time. Sessions waiting for anything else are not + terminated, nor are other processes; the operation keeps waiting for + them. A message is written to the server log for each process + whose termination is requested this way, and another if it has not + exited a few seconds later. + </para> + <para> + If this value is specified without units, it is taken as + milliseconds. A value of zero disables the timeout, so the + operation waits as long as it takes. The default is 30 seconds + (<literal>30s</literal>). This parameter can only be set in the + <filename>postgresql.conf</filename> file or on the server command + line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-max-locks-per-transaction" xreflabel="max_locks_per_transaction"> <term><varname>max_locks_per_transaction</varname> (<type>integer</type>) <indexterm> diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c index 3704d12..bc8ebe3 100644 --- a/src/backend/libpq/pqcomm.c +++ b/src/backend/libpq/pqcomm.c @@ -76,6 +76,7 @@ #include "postmaster/postmaster.h" #include "storage/ipc.h" #include "storage/latch.h" +#include "storage/proc.h" #include "utils/guc_hooks.h" #include "utils/memutils.h" @@ -1417,6 +1418,11 @@ internal_flush_buffer(const char *buf, size_t *start, size_t *end) last_reported_send_errno = 0; /* reset after any successful send */ bufptr += r; *start += r; + + /* Let WaitForProcSignalBarrier() see that we are making progress */ + if (MyProc != NULL) + pg_atomic_write_u32(&MyProc->clientSendCount, + pg_atomic_read_u32(&MyProc->clientSendCount) + 1); } *start = *end = 0; diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index 95a3269..89e2bc5 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -38,6 +38,7 @@ #include "storage/subsystems.h" #include "tcop/tcopprot.h" #include "utils/memutils.h" +#include "utils/timestamp.h" #include "utils/wait_event.h" /* @@ -119,9 +120,24 @@ const ShmemCallbacks ProcSignalShmemCallbacks = { NON_EXEC_STATIC ProcSignalHeader *ProcSignal = NULL; +/* GUC variable */ +int ProcSignalBarrierTimeout = 30000; + +/* + * A process that keeps sending to its client is terminated anyway once a + * barrier wait has lasted this many times procsignal_barrier_timeout. + */ +#define BARRIER_TIMEOUT_LIMIT_FACTOR 10 + +/* Sends to its client completed by each process, see WaitForProcSignalBarrier() */ +static uint32 *BarrierSendCounts = NULL; + static ProcSignalSlot *MyProcSignalSlot = NULL; static bool CheckProcSignal(ProcSignalReason reason); +static void SnapshotClientSendCounts(void); +static bool TerminateBarrierStraggler(int procno, uint64 generation, + int64 waited_ms, int64 idle_ms); static void CleanupProcSignalState(int status, Datum arg); static void ResetProcSignalBarrierBits(uint32 flags); static void ProcessProcSignalBarrierInternal(bool noerror); @@ -439,6 +455,8 @@ EmitProcSignalBarrier(ProcSignalBarrierType type) void WaitForProcSignalBarrier(uint64 generation) { + TimestampTz wait_start = 0; + Assert(generation <= pg_atomic_read_u64(&ProcSignal->psh_barrierGeneration)); elog(DEBUG1, @@ -450,6 +468,9 @@ WaitForProcSignalBarrier(uint64 generation) { ProcSignalSlot *slot = &ProcSignal->psh_slot[i]; uint64 oldval; + TimestampTz last_progress = 0; + TimestampTz signaled_at = 0; + bool reported_signaled = false; /* * It's important that we check only pss_barrierGeneration here and @@ -460,12 +481,110 @@ WaitForProcSignalBarrier(uint64 generation) oldval = pg_atomic_read_u64(&slot->pss_barrierGeneration); while (oldval < generation) { + long sleep_ms = 5000; + bool report_on_timeout = true; + + /* + * If the process that has not absorbed the barrier is blocked + * writing to its client and has not sent anything to it for + * procsignal_barrier_timeout, terminate it, then keep waiting for + * it to exit. A process that exits advertises PG_UINT64_MAX and + * wakes us up (see CleanupProcSignalState()). Its count of + * completed sends tells a client that is slowly reading one large + * message from one that has stopped reading. So that a client + * that reads just enough to keep that count moving cannot hold + * the barrier forever, the process is terminated anyway once the + * whole wait has lasted BARRIER_TIMEOUT_LIMIT_FACTOR times + * procsignal_barrier_timeout. A process that is waiting for + * anything else is left alone, but checked again each time we + * wake up. + * + * XXX: This applies to every caller, not only to the startup + * process. A process blocked on its client (see + * TerminateBarrierStraggler()) can hold up the end of recovery + * and the replay of WAL just as well as DROP DATABASE or + * pg_enable_data_checksums() on a primary. + */ + if (ProcSignalBarrierTimeout > 0 && signaled_at == 0) + { + TimestampTz now = GetCurrentTimestamp(); + uint32 sends; + int64 waited_ms; + int64 idle_ms; + int64 limit_ms; + + if (wait_start == 0) + { + wait_start = now; + SnapshotClientSendCounts(); + } + if (last_progress == 0) + last_progress = wait_start; + + sends = pg_atomic_read_u32(&GetPGProcByNumber(i)->clientSendCount); + if (sends != BarrierSendCounts[i]) + { + BarrierSendCounts[i] = sends; + last_progress = now; + } + + waited_ms = TimestampDifferenceMilliseconds(wait_start, now); + idle_ms = TimestampDifferenceMilliseconds(last_progress, now); + limit_ms = (int64) ProcSignalBarrierTimeout * BARRIER_TIMEOUT_LIMIT_FACTOR; + + if (idle_ms >= ProcSignalBarrierTimeout || waited_ms >= limit_ms) + { + if (TerminateBarrierStraggler(i, generation, waited_ms, idle_ms)) + signaled_at = now; + } + else + { + int64 next_ms = Min(ProcSignalBarrierTimeout - idle_ms, + limit_ms - waited_ms); + + if (next_ms < sleep_ms) + { + sleep_ms = (long) next_ms; + report_on_timeout = false; + } + } + } + if (ConditionVariableTimedSleep(&slot->pss_barrierCV, - 5000, - WAIT_EVENT_PROC_SIGNAL_BARRIER)) - ereport(LOG, - (errmsg("still waiting for backend with PID %d to accept ProcSignalBarrier", - (int) pg_atomic_read_u32(&slot->pss_pid)))); + sleep_ms, + WAIT_EVENT_PROC_SIGNAL_BARRIER) && + report_on_timeout) + { + int pid = (int) pg_atomic_read_u32(&slot->pss_pid); + + /* + * A process that is blocked writing to its client while it + * holds off interrupts, as it does while it sends an error + * message, does not act on SIGTERM until the write completes + * or fails. Say so once, instead of letting the earlier + * message suggest that it is gone. + */ + if (signaled_at != 0 && !reported_signaled) + { + PGPROC *proc = GetPGProcByNumber(i); + bool in_write; + + in_write = proc->pid == pid && + *((volatile uint32 *) &proc->wait_event_info) == WAIT_EVENT_CLIENT_WRITE; + ereport(LOG, + (errmsg("process %d has not exited %ld ms after its termination was requested", + pid, + TimestampDifferenceMilliseconds(signaled_at, + GetCurrentTimestamp())), + in_write ? + errdetail("It is still blocked sending data to its client, at a point where it cannot be interrupted, such as while sending an error message. It will exit once the client reads or the connection is closed.") : 0)); + reported_signaled = true; + } + else + ereport(LOG, + (errmsg("still waiting for backend with PID %d to accept ProcSignalBarrier", + pid))); + } oldval = pg_atomic_read_u64(&slot->pss_barrierGeneration); } ConditionVariableCancelSleep(); @@ -486,6 +605,103 @@ WaitForProcSignalBarrier(uint64 generation) pg_memory_barrier(); } +/* + * SnapshotClientSendCounts - remember how many sends to its client each + * process has completed, so that WaitForProcSignalBarrier() can tell whether + * a process that has not absorbed the barrier is still making progress. + */ +static void +SnapshotClientSendCounts(void) +{ + if (BarrierSendCounts == NULL) + BarrierSendCounts = MemoryContextAlloc(TopMemoryContext, + sizeof(uint32) * NumProcSignalSlots); + for (int i = 0; i < NumProcSignalSlots; i++) + BarrierSendCounts[i] = pg_atomic_read_u32(&GetPGProcByNumber(i)->clientSendCount); +} + +/* + * TerminateBarrierStraggler - terminate the process using ProcSignal slot + * 'procno' if it has still not absorbed barrier 'generation' and is blocked + * writing to its client. 'waited_ms' is how long the barrier wait has lasted, + * 'idle_ms' how long the process has gone without completing a send to its + * client. Returns true if it was signaled. + * + * Only a client backend or walsender waiting in ClientWrite is terminated, + * with SIGTERM, as pg_terminate_backend() would do. While blocked in a client + * write, a backend services only ProcDiePending (see + * ProcessClientWriteInterrupt()), so it cannot absorb the barrier until its + * client reads again, and SIGTERM is what releases it; a recovery conflict + * signal would not. Even SIGTERM does not, if the process is holding off + * interrupts while it writes, as when it sends an error message; our caller + * reports that if the process does not exit. + * + * Processes waiting for anything else are not signaled. A backend waiting + * for synchronous replication, which has already committed its transaction + * locally, or for more data from its client in COPY FROM STDIN absorbs the + * barrier while it waits (see ProcessProcSignalBarrierNoError()), unless a + * barrier type that may raise an ERROR is pending as well. + */ +static bool +TerminateBarrierStraggler(int procno, uint64 generation, int64 waited_ms, + int64 idle_ms) +{ + ProcSignalSlot *slot = &ProcSignal->psh_slot[procno]; + PGPROC *proc = GetPGProcByNumber(procno); + pid_t pid; + bool absorbed; + + /* + * The process that our caller saw has not absorbed the barrier may have + * exited since, and another may have taken over the slot. The PID and + * the barrier generation change together under pss_mutex, and a new + * process starts out with the current barrier generation (see + * ProcSignalInit()), so reading both under the lock tells whether the PID + * is still that of a process that has not absorbed the barrier. + */ + SpinLockAcquire(&slot->pss_mutex); + pid = (pid_t) pg_atomic_read_u32(&slot->pss_pid); + absorbed = pg_atomic_read_u64(&slot->pss_barrierGeneration) >= generation; + SpinLockRelease(&slot->pss_mutex); + + if (absorbed || pid == 0 || pid == MyProcPid) + return false; + + /* The slot is indexed by ProcNumber, so 'proc' is its owner */ + if (proc->pid != pid) + return false; + if (proc->backendType != B_BACKEND && proc->backendType != B_WAL_SENDER) + return false; + if (*((volatile uint32 *) &proc->wait_event_info) != WAIT_EVENT_CLIENT_WRITE) + return false; + + if (idle_ms >= ProcSignalBarrierTimeout) + ereport(LOG, + (errmsg("requesting termination of process %d because it has not accepted ProcSignalBarrier within %lld ms", + (int) pid, (long long) waited_ms), + errdetail("The process is blocked sending data to its client and has not sent anything for %lld ms.", + (long long) idle_ms))); + else + ereport(LOG, + (errmsg("requesting termination of process %d because it has not accepted ProcSignalBarrier within %lld ms", + (int) pid, (long long) waited_ms), + errdetail("The process is blocked sending data to its client. It is still sending, but the wait has reached its limit of %d times \"%s\".", + BARRIER_TIMEOUT_LIMIT_FACTOR, "procsignal_barrier_timeout"))); + + /* + * As in pg_signal_backend(), the process could exit and its PID be reused + * before we send the signal; that seems too unlikely to worry about. + * Signal the whole process group if we have setsid(). + */ +#ifdef HAVE_SETSID + (void) kill(-pid, SIGTERM); +#else + (void) kill(pid, SIGTERM); +#endif + + return true; +} + /* * Handle receipt of an interrupt indicating a global barrier event. * diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index a168bac..f721623 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -372,6 +372,7 @@ ProcGlobalShmemInit(void *arg) pg_atomic_init_u32(&(proc->procArrayGroupNext), INVALID_PROC_NUMBER); pg_atomic_init_u32(&(proc->clogGroupNext), INVALID_PROC_NUMBER); pg_atomic_init_u64(&(proc->waitStart), 0); + pg_atomic_init_u32(&(proc->clientSendCount), 0); } /* Should have consumed exactly the expected amount of fast-path memory. */ diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 380679a..c579b07 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -2423,6 +2423,16 @@ check_hook => 'check_primary_slot_name', }, +{ name => 'procsignal_barrier_timeout', type => 'int', context => 'PGC_SIGHUP', group => 'LOCK_MANAGEMENT', + short_desc => 'Sets the time to wait for a session to accept a global state change before terminating it.', + long_desc => '0 disables the timeout.', + flags => 'GUC_UNIT_MS', + variable => 'ProcSignalBarrierTimeout', + boot_val => '30000', + min => '0', + max => 'INT_MAX', +}, + { name => 'quote_all_identifiers', type => 'bool', context => 'PGC_USERSET', group => 'COMPAT_OPTIONS_PREVIOUS', short_desc => 'When generating SQL fragments, quote all identifiers.', variable => 'quote_all_identifiers', diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 342aaee..8c03a4f 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -88,6 +88,7 @@ #include "storage/predicate.h" #include "storage/proc.h" #include "storage/procnumber.h" +#include "storage/procsignal.h" #include "storage/standby.h" #include "tcop/backend_startup.h" #include "tcop/tcopprot.h" diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 01f98de..9786f25 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -876,6 +876,9 @@ #------------------------------------------------------------------------------ #deadlock_timeout = 1s +#procsignal_barrier_timeout = 30s # terminate a session that has not + # accepted a global state change + # after this long; 0 disables #max_locks_per_transaction = 128 # min 10 # (change requires restart) #max_pred_locks_per_transaction = 64 # min 10 diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index abe4000..3892f94 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -383,6 +383,13 @@ typedef struct PGPROC ************************************************************************/ uint32 wait_event_info; /* proc's wait information */ + + /* + * Number of sends to the client completed by this process. Written only + * by the process itself; read by WaitForProcSignalBarrier() to tell a + * slow client from one that has stopped reading. + */ + pg_atomic_uint32 clientSendCount; } PGPROC; diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index 7721814..f4e5ad5 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -71,6 +71,9 @@ typedef enum */ #define MAX_CANCEL_KEY_LENGTH 32 +/* GUC variable */ +extern PGDLLIMPORT int ProcSignalBarrierTimeout; + /* * prototypes for functions in procsignal.c */ diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 5732992..4325c5c 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -70,6 +70,7 @@ tests += { 't/059_remote_apply_status_interval.pl', 't/060_multixact_crashed_updater.pl', 't/061_procsignal_barrier_absorb.pl', + 't/062_procsignal_barrier_timeout.pl', ], }, } diff --git a/src/test/recovery/t/062_procsignal_barrier_timeout.pl b/src/test/recovery/t/062_procsignal_barrier_timeout.pl new file mode 100644 index 0000000..e20bad1 --- /dev/null +++ b/src/test/recovery/t/062_procsignal_barrier_timeout.pl @@ -0,0 +1,526 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test procsignal_barrier_timeout. Operations that wait for every process to +# absorb a ProcSignalBarrier must not wait forever behind a session that is +# blocked writing to a client that has stopped reading: after the timeout such +# a session is terminated. Sessions that are merely idle, sleeping or reading +# slowly must not be, nor a session sending a large value to a client that +# keeps reading it, until the operation has waited ten times the timeout. +# Sessions waiting for synchronous replication or for more COPY FROM STDIN data +# absorb the barrier while they wait (see 061_procsignal_barrier_absorb.pl). A session blocked sending an +# error message cannot be terminated; the operation says so and keeps waiting. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; +use Time::HiRes qw(time usleep); + +my $copy_query = + "COPY (SELECT generate_series(1, 1000000000), repeat('x', 200)) TO STDOUT"; +my $terminated_re = + 'because it has not accepted ProcSignalBarrier within \d+ ms'; +my $timeout = $PostgreSQL::Test::Utils::timeout_default; + +# Start a psql running $sql, with its output captured but not read. +sub start_psql +{ + my ($node, $dbname, $appname, $sql, $out) = @_; + my $err = ''; + + return IPC::Run::start( + [ + 'psql', '--no-psqlrc', '--quiet', + '--dbname' => $node->connstr($dbname) + . " application_name=$appname", + '--command' => $sql + ], + '>' => $out, + '2>' => \$err); +} + +# Return the PID of the session with the given application_name, once its +# wait_event is $wait_event (or once it exists, if $wait_event is undef). +sub session_pid +{ + my ($node, $appname, $wait_event) = @_; + my $cond = defined $wait_event ? "wait_event = '$wait_event'" : 'true'; + + $node->poll_query_until('postgres', + "SELECT $cond FROM pg_stat_activity WHERE application_name = '$appname'" + ) or die "timed out waiting for session \"$appname\""; + return $node->safe_psql('postgres', + "SELECT pid FROM pg_stat_activity WHERE application_name = '$appname'"); +} + +# Start a COPY TO STDOUT whose output is never read. psql blocks writing to +# its stdout and stops reading from the server, so the backend ends up blocked +# in a client write, where it cannot absorb a ProcSignalBarrier. +sub start_stuck_session +{ + my ($node, $dbname, $appname) = @_; + my $out = ''; + my $h = start_psql($node, $dbname, $appname, $copy_query, \$out); + my $pid = session_pid($node, $appname, 'ClientWrite'); + + wait_until_stuck($node, $pid); + return ($h, $pid); +} + +# Wait until the session $pid, which shows ClientWrite, is blocked for good. It +# can show ClientWrite while psql is still reading, and even once psql has +# stopped, a backend that is woken up can find room for one more send on a +# Unix-domain socket and run on to its next write, absorbing any pending +# barrier on the way. So wake it up until it stays put: +# pg_log_backend_memory_contexts() makes it log once it reaches its next +# CHECK_FOR_INTERRUPTS(), which it only does if it was not blocked for good. +sub wait_until_stuck +{ + my ($node, $pid) = @_; + + for (my $i = 0;; $i++) + { + my $offset = -s $node->logfile; + + die "session $pid does not stay blocked" if $i >= 20; + $node->safe_psql('postgres', + "SELECT pg_log_backend_memory_contexts($pid)"); + usleep(500_000); + last + unless $node->log_contains("logging memory contexts of PID $pid", + $offset); + $node->poll_query_until('postgres', + "SELECT wait_event = 'ClientWrite' FROM pg_stat_activity WHERE pid = $pid" + ) or die "timed out waiting for session $pid to block again"; + } +} + +sub session_alive +{ + my ($node, $pid) = @_; + + return $node->safe_psql('postgres', + "SELECT count(*) FROM pg_stat_activity WHERE pid = $pid") eq '1'; +} + +# Open a connection by hand and send $sql as a simple query, without reading +# the reply. psql hands a row over only once it has received all of it, so it +# cannot read a single large message slowly; this can, or not read it at all. +# $options are passed to the server as for PGOPTIONS. +sub start_raw_session +{ + my ($node, $appname, $sql, $options) = @_; + my $user = $node->safe_psql('postgres', 'SELECT current_user'); + my $sock = $node->raw_connect(); + my $params = + "user\0$user\0database\0postgres\0application_name\0$appname\0"; + my ($type, $buf); + + $params .= "options\0$options\0" if defined $options; + $params .= "\0"; + $sock->syswrite(pack('NN', 8 + length($params), 196608) . $params); + # Read messages until ReadyForQuery + do + { + $type = raw_read($sock, 1); + $buf = raw_read($sock, 4); + raw_read($sock, unpack('N', $buf) - 4); + } while ($type ne 'Z'); + $sock->syswrite('Q' . pack('N', 5 + length($sql)) . $sql . "\0"); + return $sock; +} + +# Read exactly $len bytes from $sock. +sub raw_read +{ + my ($sock, $len) = @_; + my $buf = ''; + + while (length($buf) < $len) + { + my $n = $sock->sysread($buf, $len - length($buf), length($buf)); + + die "connection closed" if !$n; + } + return $buf; +} + +# Run $sql in the background while reading from $sock, 8kB at a time with a +# pause in between, so that the session sends a large message slowly but +# steadily. Returns the time $sql took, once it has completed and either +# $sock has received $until or the connection was closed. +sub run_while_reading_raw_slowly +{ + my ($node, $sock, $sql, $until) = @_; + my $out = ''; + my $t0 = time; + my $op = start_psql($node, 'postgres', 'op_raw', $sql, \$out); + my ($op_elapsed, $tail, $closed) = (undef, '', 0); + + while (1) + { + if (!$closed && $tail ne $until) + { + my $buf = ''; + my $n = $sock->sysread($buf, 8192); + + $closed = 1 if !$n; + $tail = substr($tail . $buf, -length($until)) if $n; + } + if (!defined $op_elapsed) + { + if ($op->pumpable) + { + $op->pump_nb; + } + else + { + $op->finish; + $op_elapsed = time - $t0; + } + } + last if defined $op_elapsed && ($closed || $tail eq $until); + die "timed out running \"$sql\"" if time - $t0 > $timeout; + usleep(10_000) if !defined $op_elapsed; + } + return $op_elapsed; +} + +# Check that the session $pid was terminated by the barrier timeout. +sub check_terminated +{ + my ($node, $pid, $log_offset, $what) = @_; + + ok( $node->log_contains( + "requesting termination of process $pid $terminated_re", + $log_offset), + "$what: blocked session was terminated after the timeout"); + ok( $node->log_contains( + qr/\[$pid\].*FATAL: terminating connection due to administrator command/, + $log_offset), + "$what: blocked session exited"); + ok($node->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE pid = $pid"), + "$what: blocked session is gone"); +} + +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1, no_data_checksums => 1); +$primary->append_conf('postgresql.conf', 'procsignal_barrier_timeout = 2s'); +$primary->start; +$primary->safe_psql('postgres', 'CREATE DATABASE other'); +$primary->safe_psql('postgres', 'CREATE DATABASE target1'); +$primary->safe_psql('postgres', 'CREATE DATABASE target2'); +$primary->safe_psql('postgres', 'CREATE DATABASE target3'); +$primary->safe_psql('postgres', 'CREATE DATABASE target4'); +$primary->safe_psql('postgres', 'CREATE DATABASE target7'); +$primary->safe_psql('postgres', 'CREATE DATABASE target8'); +$primary->safe_psql('postgres', 'CREATE DATABASE target9'); + +my ($h, $pid, $log_offset, $start, $elapsed); + +# +# DROP DATABASE waits for every process to close its files of the dropped +# database, including a session connected to another database. +# +($h, $pid) = start_stuck_session($primary, 'other', 'stuck_dropdb'); +$log_offset = -s $primary->logfile; +$start = time; +$primary->safe_psql('postgres', 'DROP DATABASE target1', + timeout => $timeout); +$elapsed = time - $start; +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target1'"), + '0', + 'DROP DATABASE: completed with a blocked session on another database'); +cmp_ok($elapsed, '>=', 2, 'DROP DATABASE: waited for the timeout'); +check_terminated($primary, $pid, $log_offset, 'DROP DATABASE'); +$h->kill_kill; + +# +# Enabling data checksums online. +# +($h, $pid) = start_stuck_session($primary, 'postgres', 'stuck_checksums'); +$log_offset = -s $primary->logfile; +$primary->safe_psql('postgres', 'SELECT pg_enable_data_checksums()'); +ok( $primary->poll_query_until( + 'postgres', + "SELECT setting FROM pg_settings WHERE name = 'data_checksums'", 'on'), + 'pg_enable_data_checksums(): reached "on" with a blocked session'); +check_terminated($primary, $pid, $log_offset, 'pg_enable_data_checksums()'); +$h->kill_kill; + +# +# A WAL sender blocked sending a base backup to a client that has stopped +# reading. +# +my ($bb_out, $bb_err) = ('', ''); +$h = IPC::Run::start( + [ + 'pg_basebackup', + '--pgdata' => '-', + '--format' => 'tar', + '--wal-method' => 'none', + '--checkpoint' => 'fast', + '--dbname' => $primary->connstr('postgres') + . ' application_name=stuck_walsender' + ], + '>' => \$bb_out, + '2>' => \$bb_err); +$pid = session_pid($primary, 'stuck_walsender', 'ClientWrite'); +wait_until_stuck($primary, $pid); +is( $primary->safe_psql( + 'postgres', "SELECT backend_type FROM pg_stat_activity WHERE pid = $pid"), + 'walsender', + 'base backup is served by a WAL sender'); +$log_offset = -s $primary->logfile; +$primary->safe_psql('postgres', 'DROP DATABASE target4', + timeout => $timeout); +check_terminated($primary, $pid, $log_offset, 'WAL sender'); +$h->kill_kill; + +# +# A session blocked while it sends an error message to a client that has +# stopped reading holds off interrupts, so SIGTERM does not release it. The +# operation says that the session has not exited and keeps waiting for it, as +# it would without the timeout, until the client goes away. +# +my $raw = start_raw_session($primary, 'stuck_error', + "SELECT repeat('x', 10000000)::int", + '-c log_min_messages=fatal'); +$pid = session_pid($primary, 'stuck_error', 'ClientWrite'); +wait_until_stuck($primary, $pid); +$log_offset = -s $primary->logfile; +my $err_dropper = $primary->background_psql('postgres'); +$err_dropper->query_until(qr/dropping/, + "\\echo dropping\nDROP DATABASE target7;\n"); +$primary->wait_for_log( + qr/requesting termination of process $pid $terminated_re/, + $log_offset); +$primary->wait_for_log( + qr/process $pid has not exited \d+ ms after its termination was requested/, + $log_offset); +ok(session_alive($primary, $pid), + 'error message: blocked session has not exited'); +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target7'"), + '1', + 'error message: DROP DATABASE is still waiting'); +$raw->close; +$err_dropper->quit; +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target7'"), + '0', + 'error message: DROP DATABASE completed once the client went away'); +ok( $primary->log_contains( + qr/\[$pid\].*FATAL: terminating connection due to administrator command/, + $log_offset), + 'error message: blocked session exited'); + +# +# A session sending one large value to a client that keeps reading it slowly +# cannot absorb the barrier before the whole value is out either, but it keeps +# completing sends, so it is not terminated: the operation waits for the value +# to be out, as it would without the timeout. +# +$raw = start_raw_session($primary, 'slow_bigvalue', + "SELECT repeat('x', 6000000)"); +$pid = session_pid($primary, 'slow_bigvalue', 'ClientWrite'); +$log_offset = -s $primary->logfile; +$elapsed = run_while_reading_raw_slowly($primary, $raw, + 'DROP DATABASE target8', "Z\0\0\0\5I"); +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target8'"), + '0', + 'large value: DROP DATABASE completed'); +cmp_ok($elapsed, '>=', 2, + 'large value: DROP DATABASE waited longer than the timeout'); +ok(!$primary->log_contains('requesting termination', $log_offset), + 'large value: no session was terminated'); +ok(session_alive($primary, $pid), + 'large value: slowly reading session survived'); +$raw->close; + +# +# But a session that is still sending is terminated anyway once the operation +# has waited ten times the timeout, so that a client cannot hold it up forever +# by reading just enough. +# +$raw = start_raw_session($primary, 'slower_bigvalue', + "SELECT repeat('x', 40000000)"); +$pid = session_pid($primary, 'slower_bigvalue', 'ClientWrite'); +$log_offset = -s $primary->logfile; +$elapsed = run_while_reading_raw_slowly($primary, $raw, + 'DROP DATABASE target9', "Z\0\0\0\5I"); +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target9'"), + '0', + 'still sending: DROP DATABASE completed'); +cmp_ok($elapsed, '>=', 20, + 'still sending: DROP DATABASE waited ten times the timeout'); +ok( $primary->log_contains( + qr/requesting termination of process $pid $terminated_re.*\n.*It is still sending, but the wait has reached its limit/, + $log_offset), + 'still sending: session was terminated at the limit'); +ok($primary->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE pid = $pid"), + 'still sending: session is gone'); +$raw->close; + +# +# Negative control: idle sessions, a session in pg_sleep() and a client that +# reads slowly but keeps reading are not terminated, and do not delay the +# operations. +# +my $idle = $primary->background_psql('postgres'); +my $idle_pid = $idle->query_safe('SELECT pg_backend_pid()'); +my $sleep_out = ''; +my $sleeper = start_psql($primary, 'postgres', 'sleeper', + 'SELECT pg_sleep(600)', \$sleep_out); +my $sleeper_pid = session_pid($primary, 'sleeper', 'PgSleep'); +my $slow_out = ''; +my $slow = start_psql($primary, 'other', 'slow_reader', $copy_query, + \$slow_out); +my $slow_pid = session_pid($primary, 'slow_reader', 'ClientWrite'); + +# Run $sql, reading a little from the slow client now and then until $sql has +# completed and the SQL condition $done holds. The slow client must keep +# reading for as long as a barrier may be waiting for it: +# pg_disable_data_checksums() returns before the data checksums launcher +# emits and waits for its barriers. Returns the time $sql took. +sub run_while_reading_slowly +{ + my ($sql, $done) = @_; + my $out = ''; + my $t0 = time; + my $op = start_psql($primary, 'postgres', 'op', $sql, \$out); + my $op_elapsed; + + while (1) + { + $slow->pump_nb; + $slow_out = ''; + if (!defined $op_elapsed) + { + if ($op->pumpable) + { + $op->pump_nb; + } + else + { + $op->finish; + $op_elapsed = time - $t0; + } + } + last + if defined $op_elapsed + && $primary->safe_psql('postgres', "SELECT $done") eq 't'; + die "timed out running \"$sql\"" if time - $t0 > $timeout; + usleep(50_000); + } + return $op_elapsed; +} + +$log_offset = -s $primary->logfile; +$elapsed = run_while_reading_slowly('DROP DATABASE target2', + "NOT EXISTS (SELECT FROM pg_database WHERE datname = 'target2')"); +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target2'"), + '0', + 'negative control: DROP DATABASE completed'); +cmp_ok($elapsed, '<', 2, + 'negative control: DROP DATABASE did not wait for the timeout'); +run_while_reading_slowly('SELECT pg_disable_data_checksums()', + "(SELECT setting FROM pg_settings WHERE name = 'data_checksums') = 'off'"); +pass('negative control: data checksums disabled'); +ok(!$primary->log_contains('requesting termination', $log_offset), + 'negative control: no session was terminated'); +ok( session_alive($primary, $idle_pid) + && session_alive($primary, $sleeper_pid) + && session_alive($primary, $slow_pid), + 'negative control: idle, sleeping and slowly reading sessions survived'); +$idle->quit; +$sleeper->kill_kill; +$slow->kill_kill; + +# +# With procsignal_barrier_timeout = 0, the operation keeps waiting for the +# blocked session. +# +$primary->append_conf('postgresql.conf', 'procsignal_barrier_timeout = 0'); +$primary->reload; +$primary->poll_query_until('postgres', 'SHOW procsignal_barrier_timeout', + '0') + or die 'timed out waiting for the reload'; +($h, $pid) = start_stuck_session($primary, 'other', 'stuck_disabled'); +$log_offset = -s $primary->logfile; +my $dropper = $primary->background_psql('postgres'); +$dropper->query_until(qr/dropping/, "\\echo dropping\nDROP DATABASE target3;\n"); +$primary->wait_for_log( + qr/still waiting for backend with PID $pid to accept ProcSignalBarrier/, + $log_offset); +ok(session_alive($primary, $pid), + 'timeout disabled: blocked session is not terminated'); +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target3'"), + '1', + 'timeout disabled: DROP DATABASE is still waiting'); +$primary->safe_psql('postgres', "SELECT pg_terminate_backend($pid)"); +$dropper->quit; +is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = 'target3'"), + '0', + 'timeout disabled: DROP DATABASE completed once the session was terminated' +); +$h->kill_kill; +$primary->append_conf('postgresql.conf', 'procsignal_barrier_timeout = 2s'); +$primary->reload; + +# +# Standby: replay of the record that enables logical decoding with +# wal_level = replica, and the end of recovery when logical decoding is +# enabled, wait for every process to update its cached state. +# +$primary->backup('backup'); +my $standby = PostgreSQL::Test::Cluster->new('standby'); +$standby->init_from_backup($primary, 'backup', has_streaming => 1); +$standby->append_conf('postgresql.conf', 'procsignal_barrier_timeout = 2s'); +$standby->start; +$primary->wait_for_replay_catchup($standby); + +($h, $pid) = start_stuck_session($standby, 'postgres', 'stuck_replay'); +$log_offset = -s $standby->logfile; +$primary->safe_psql('postgres', + "SELECT pg_create_logical_replication_slot('primary_slot', 'pgoutput')"); +$primary->wait_for_replay_catchup($standby); +is($standby->safe_psql('postgres', 'SHOW effective_wal_level'), + 'logical', 'replay: logical decoding enabled with a blocked session'); +check_terminated($standby, $pid, $log_offset, 'replay'); +$h->kill_kill; + +$standby->create_logical_slot_on_standby($primary, 'standby_slot', + 'postgres'); +$idle = $standby->background_psql('postgres'); +$idle_pid = $idle->query_safe('SELECT pg_backend_pid()'); +($h, $pid) = start_stuck_session($standby, 'postgres', 'stuck_promote'); +$log_offset = -s $standby->logfile; +$standby->promote; +is($standby->safe_psql('postgres', 'SELECT pg_is_in_recovery()'), + 'f', 'promotion: completed with a blocked session'); +check_terminated($standby, $pid, $log_offset, 'promotion'); +ok(session_alive($standby, $idle_pid), + 'promotion: idle session survived'); +$idle->quit; +$h->kill_kill; + +done_testing(); -- 2.56.0
