From 80603570e01b44bd0dad1c4ba2db105ebf2e5fc9 Mon Sep 17 00:00:00 2001 From: Manu Date: Fri, 23 Oct 2026 01:06:20 -0300 Subject: [PATCH v3] Terminate sessions that do not absorb a ProcSignalBarrier in time 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, even if it cannot absorb the barrier while it waits either: a backend waiting for synchronous replication has already committed locally, and one in COPY FROM STDIN is waiting for its client to send more data. 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 Suggested-by: Masahiko Sawada Suggested-by: shihao zhong Reviewed-by: shihao zhong Discussion: https://postgr.es/m/CAA4eK1LC09Tq3vTr04ppXRbyQnMb66_pgyuES2C_5ZoWfYRUrQ@mail.gmail.com --- doc/src/sgml/config.sgml | 44 ++ 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/061_procsignal_barrier_timeout.pl | 615 ++++++++++++++++++ 11 files changed, 912 insertions(+), 5 deletions(-) create mode 100644 src/test/recovery/t/061_procsignal_barrier_timeout.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index f7db299..61f8af2 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11836,6 +11836,50 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' + + procsignal_barrier_timeout (integer) + + procsignal_barrier_timeout configuration parameter + + + + + 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 + ) and + DROP DATABASE. 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 + pg_terminate_backend 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 + anything else, for example for synchronous replication or for more + data from the client during COPY FROM STDIN, 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. + + + 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 + (30s). This parameter can only be set in the + postgresql.conf file or on the server command + line. + + + + max_locks_per_transaction (integer) 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 eb0f4d6..96aee4c 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); @@ -437,6 +453,8 @@ EmitProcSignalBarrier(ProcSignalBarrierType type) void WaitForProcSignalBarrier(uint64 generation) { + TimestampTz wait_start = 0; + Assert(generation <= pg_atomic_read_u64(&ProcSignal->psh_barrierGeneration)); elog(DEBUG1, @@ -448,6 +466,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 @@ -458,12 +479,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(); @@ -484,6 +603,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, even when they cannot + * absorb the barrier either while they wait. A backend waiting for + * synchronous replication has already committed its transaction locally (see + * SyncRepWaitForLSN()), and one in COPY FROM STDIN is waiting for its client + * to send more data, which the client may be about to do. + */ +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 aaa158b..9f048c2 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -66,6 +66,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 b6ac9dd..4942a0c 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_timeout.pl', ], }, } diff --git a/src/test/recovery/t/061_procsignal_barrier_timeout.pl b/src/test/recovery/t/061_procsignal_barrier_timeout.pl new file mode 100644 index 0000000..2183325 --- /dev/null +++ b/src/test/recovery/t/061_procsignal_barrier_timeout.pl @@ -0,0 +1,615 @@ +# 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 sessions that wait for something other than their +# client reading, such as synchronous replication or more COPY FROM STDIN data, +# nor a session sending a large value to a client that keeps reading it, until +# the operation has waited ten times the timeout. 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 target5'); +$primary->safe_psql('postgres', 'CREATE DATABASE target6'); +$primary->safe_psql('postgres', 'CREATE DATABASE target7'); +$primary->safe_psql('postgres', 'CREATE DATABASE target8'); +$primary->safe_psql('postgres', 'CREATE DATABASE target9'); +$primary->safe_psql('postgres', 'CREATE TABLE copy_target (a int)'); + +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; + +# +# Sessions that cannot absorb the barrier while they wait, but are not blocked +# writing to their client, are not terminated either. A session waiting for +# synchronous replication has already committed locally, and one in COPY FROM +# STDIN waits for its client to send more data. The operation keeps waiting +# for them and completes once they move on. +# +sub check_not_terminated +{ + my ($pid, $dbname, $what, $release) = @_; + my $offset = -s $primary->logfile; + my $dropper = $primary->background_psql('postgres'); + + $dropper->query_safe('SET synchronous_commit = local'); + $dropper->query_until(qr/dropping/, + "\\echo dropping\nDROP DATABASE $dbname;\n"); + + # Reported only once the timeout has expired and the session was left + # alone. + $primary->wait_for_log( + qr/still waiting for backend with PID $pid to accept ProcSignalBarrier/, + $offset); + ok(session_alive($primary, $pid), "$what: session is not terminated"); + ok(!$primary->log_contains('requesting termination', $offset), + "$what: no session was terminated"); + is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = '$dbname'"), + '1', + "$what: DROP DATABASE is still waiting"); + + $release->(); + $dropper->quit; + is( $primary->safe_psql( + 'postgres', + "SELECT count(*) FROM pg_database WHERE datname = '$dbname'"), + '0', + "$what: DROP DATABASE completed once the session moved on"); +} + +# A session waiting for a synchronous standby that is not connected. +$primary->safe_psql('postgres', + "ALTER SYSTEM SET synchronous_standby_names = 'nosuch'"); +$primary->reload; +my $syncrep_out = ''; +my $syncrep = start_psql($primary, 'postgres', 'syncrep_waiter', + 'CREATE TABLE syncrep_target ()', \$syncrep_out); +my $syncrep_pid = session_pid($primary, 'syncrep_waiter', 'SyncRep'); +check_not_terminated( + $syncrep_pid, + 'target5', + 'synchronous replication', + sub { + $primary->safe_psql('postgres', + 'ALTER SYSTEM RESET synchronous_standby_names'); + $primary->reload; + }); +$syncrep->finish; + +# 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 = IPC::Run::start( + [ + 'psql', '--no-psqlrc', '--quiet', + '--dbname' => $primary->connstr('postgres') + . ' application_name=copy_paused' + ], + '<' => \$copy_in, + '>' => \$copy_out, + '2>' => \$copy_err); +$copy->pump_nb; +$primary->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($primary, 'copy_paused'); +check_not_terminated( + $copy_pid, + 'target6', + 'COPY FROM STDIN', + sub { + $copy_in .= "2\n\\.\n"; + $copy->finish; + }); +is($primary->safe_psql('postgres', 'SELECT count(*) FROM copy_target'), + '2', 'COPY FROM STDIN: completed'); + +# +# 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