From 05da6b0dcf5a9255ee62a0bac667f3a13366502d Mon Sep 17 00:00:00 2001 From: Hari Babu Date: Thu, 28 Mar 2019 15:30:01 +1100 Subject: [PATCH 8/8] Server recovery mode handling in_recovery GUC_REPORT is added to update the clients when the server is recovery mode, this is useful for the client connections to connect to a standby server with a faster check instead of executing a command. New SIGUSR1 handling interrupt is added to support reporting of recovery mode exit to all backends and their respective clients. Some parts of the code is taken from earlier development by Elvis Pranskevichus and Tsunakawa Takayuki. --- doc/src/sgml/libpq.sgml | 14 ++- doc/src/sgml/protocol.sgml | 8 +- src/backend/access/transam/xlog.c | 3 + src/backend/commands/async.c | 1 - src/backend/storage/ipc/procarray.c | 28 ++++++ src/backend/storage/ipc/procsignal.c | 3 + src/backend/storage/ipc/standby.c | 9 ++ src/backend/tcop/postgres.c | 60 ++++++++++++ src/backend/utils/init/postinit.c | 6 +- src/backend/utils/misc/check_guc | 2 +- src/backend/utils/misc/guc.c | 17 +++- src/include/storage/procarray.h | 1 + src/include/storage/procsignal.h | 2 + src/include/storage/standby.h | 1 + src/include/tcop/tcopprot.h | 2 + src/interfaces/libpq/fe-connect.c | 133 +++++++++++++++++---------- src/interfaces/libpq/fe-exec.c | 4 + src/interfaces/libpq/libpq-int.h | 1 + 18 files changed, 234 insertions(+), 61 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index 6b84accd08..90e8c96820 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -1702,8 +1702,10 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname To find out whether the server is in recovery mode or not, query SELECT pg_is_in_recovery() - will be sent upon any successful connection; if it returns t, means server - is in recovery mode. + will be sent upon any successful connection if the server is prior to version 12; if it returns + t, it means server is in recovery mode. But for server version 12 or greater + uses the value of in_recovery configuration parameter that is reported by the + server upon successful connection. @@ -2071,15 +2073,17 @@ const char *PQparameterStatus(const PGconn *conn, const char *paramName); IntervalStyle, TimeZone, integer_datetimes, - standard_conforming_strings, and - transaction_read_only. + standard_conforming_strings, + transaction_read_only and + in_recovery. (server_encoding, TimeZone, and integer_datetimes were not reported by releases before 8.0; standard_conforming_strings was not reported by releases before 8.1; IntervalStyle was not reported by releases before 8.4; application_name was not reported by releases before 9.0; - transaction_read_only was not reported by release before 12.0.) + transaction_read_only and in_recovery + were not reported by release before 12.0.) Note that server_version, server_encoding and diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 05286c7365..9ed2a03b73 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -1284,15 +1284,17 @@ SELECT 1/0; IntervalStyle, TimeZone, integer_datetimes, - standard_conforming_strings, and - transaction_read_only. + standard_conforming_strings, + transaction_read_only and + in_recovery. (server_encoding, TimeZone, and integer_datetimes were not reported by releases before 8.0; standard_conforming_strings was not reported by releases before 8.1; IntervalStyle was not reported by releases before 8.4; application_name was not reported by releases before 9.0; - transaction_read_only was not reported by releases before 12.0.) + transaction_read_only and in_recovery + were not reported by releases before 12.0.) Note that server_version, server_encoding and diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 1c7dd51b9f..976e6f63ae 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7755,6 +7755,9 @@ StartupXLOG(void) XLogCtl->SharedRecoveryInProgress = false; SpinLockRelease(&XLogCtl->info_lck); + if (standbyState != STANDBY_DISABLED) + SendRecoveryExitSignal(); + UpdateControlFile(); LWLockRelease(ControlFileLock); diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c index 738e6ec7e2..102094e765 100644 --- a/src/backend/commands/async.c +++ b/src/backend/commands/async.c @@ -1736,7 +1736,6 @@ ProcessNotifyInterrupt(void) ProcessIncomingNotify(); } - /* * Read all pending notifications from the queue, and deliver appropriate * ones to my frontend. Stop when we reach queue head or an uncommitted diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 18a0f62ba6..ad72ab10c7 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -2972,6 +2972,34 @@ CountOtherDBBackends(Oid databaseId, int *nbackends, int *nprepared) return true; /* timed out, still conflicts */ } +/* + * SendSignalToAllBackends --- send a signal to all backends. + */ +void +SendSignalToAllBackends(ProcSignalReason reason) +{ + ProcArrayStruct *arrayP = procArray; + int index; + pid_t pid = 0; + + LWLockAcquire(ProcArrayLock, LW_SHARED); + + for (index = 0; index < arrayP->numProcs; index++) + { + int pgprocno = arrayP->pgprocnos[index]; + volatile PGPROC *proc = &allProcs[pgprocno]; + VirtualTransactionId procvxid; + + GET_VXID_FROM_PGPROC(procvxid, *proc); + + pid = proc->pid; + if (pid != 0) + (void) SendProcSignal(pid, reason, procvxid.backendId); + } + + LWLockRelease(ProcArrayLock); +} + /* * ProcArraySetReplicationSlotXmin * diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c index 7605b2c367..e4548dc323 100644 --- a/src/backend/storage/ipc/procsignal.c +++ b/src/backend/storage/ipc/procsignal.c @@ -292,6 +292,9 @@ procsignal_sigusr1_handler(SIGNAL_ARGS) if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN); + if (CheckProcSignal(PROCSIG_RECOVERY_EXIT)) + HandleRecoveryExitInterrupt(); + SetLatch(MyLatch); latch_sigusr1_handler(); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 842fcabd97..831cd80357 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -138,6 +138,15 @@ ShutdownRecoveryTransactionEnvironment(void) VirtualXactLockTableCleanup(); } +/* + * SendRecoveryExitSignal + * Signal backends that the server has exited recovery mode. + */ +void +SendRecoveryExitSignal(void) +{ + SendSignalToAllBackends(PROCSIG_RECOVERY_EXIT); +} /* * ----------------------------------------------------- diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 44a59e1d4f..33511a9433 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -167,6 +167,15 @@ static bool RecoveryConflictPending = false; static bool RecoveryConflictRetryable = true; static ProcSignalReason RecoveryConflictReason; +/* + * Inbound recovery exit are initially processed by + * HandleRecoveryExitInterrupt(), called from inside a signal handler. + * That just sets the recoveryExitInterruptPending flag and sets the process + * latch. ProcessRecoveryExitInterrupt() will then be called whenever it's + * safe to actually deal with the interrupt. + */ +volatile sig_atomic_t recoveryExitInterruptPending = false; + /* reused buffer to pass to SendRowDescriptionMessage() */ static MemoryContext row_description_context = NULL; static StringInfoData row_description_buf; @@ -195,6 +204,7 @@ static void drop_unnamed_stmt(void); static void log_disconnections(int code, Datum arg); static void enable_statement_timeout(void); static void disable_statement_timeout(void); +static void ProcessRecoveryExitInterrupt(void); /* ---------------------------------------------------------------- @@ -543,6 +553,10 @@ ProcessClientReadInterrupt(bool blocked) /* Process notify interrupts, if any */ if (notifyInterruptPending) ProcessNotifyInterrupt(); + + /* Process recovery exit interrupts that happened while reading */ + if (recoveryExitInterruptPending) + ProcessRecoveryExitInterrupt(); } else if (ProcDiePending) { @@ -2954,6 +2968,52 @@ RecoveryConflictInterrupt(ProcSignalReason reason) errno = save_errno; } +/* + * HandleRecoveryExitInterrupt + * + * Signal handler portion of interrupt handling. Let the backend know + * that the server has exited the recovery mode. + */ +void +HandleRecoveryExitInterrupt(void) +{ + /* + * Note: this is called by a SIGNAL HANDLER. You must be very wary what + * you do here. + */ + + /* signal that work needs to be done */ + recoveryExitInterruptPending = true; + + /* make sure the event is processed in due course */ + SetLatch(MyLatch); +} + +/* + * ProcessRecoveryExitInterrupt + * + * This is called just after waiting for a frontend command. If a + * interrupt arrives (via HandleRecoveryExitInterrupt()) while reading, + * the read will be interrupted via the process's latch, and this routine + * will get called. +*/ +static void +ProcessRecoveryExitInterrupt(void) +{ + recoveryExitInterruptPending = false; + + SetConfigOption("in_recovery", + "off", + PGC_INTERNAL, PGC_S_OVERRIDE); + + /* + * Flush output buffer so that clients receive the ParameterStatus message + * as soon as possible. + */ + pq_flush(); +} + + /* * ProcessInterrupts: out-of-line portion of CHECK_FOR_INTERRUPTS() macro * diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c index e9f72b5069..29a159dc11 100644 --- a/src/backend/utils/init/postinit.c +++ b/src/backend/utils/init/postinit.c @@ -651,7 +651,11 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username, * This is handled by calling RecoveryInProgress and ignoring the * result. */ - (void) RecoveryInProgress(); + if (RecoveryInProgress()) + SetConfigOption("in_recovery", + "on", + PGC_INTERNAL, PGC_S_OVERRIDE); + } else { diff --git a/src/backend/utils/misc/check_guc b/src/backend/utils/misc/check_guc index d228bbed68..1c0f51ac8b 100755 --- a/src/backend/utils/misc/check_guc +++ b/src/backend/utils/misc/check_guc @@ -21,7 +21,7 @@ is_superuser lc_collate lc_ctype lc_messages lc_monetary lc_numeric lc_time \ pre_auth_delay role seed server_encoding server_version server_version_int \ session_authorization trace_lock_oidmin trace_lock_table trace_locks trace_lwlocks \ trace_notify trace_userlocks transaction_isolation transaction_read_only \ -zero_damaged_pages" +zero_damaged_pages in_recovery" ### What options are listed in postgresql.conf.sample, but don't appear ### in guc.c? diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index e86b4d3319..23b5b3e46c 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -582,7 +582,7 @@ static char *recovery_target_xid_string; static char *recovery_target_time_string; static char *recovery_target_name_string; static char *recovery_target_lsn_string; - +static bool in_recovery; /* should be static, but commands/variable.c needs to get at this */ char *role_string; @@ -1771,6 +1771,21 @@ static struct config_bool ConfigureNamesBool[] = NULL, NULL, NULL }, + { + /* + * Not for general use --- used to indicate whether the instance is + * recovery mode + */ + {"in_recovery", PGC_INTERNAL, UNGROUPED, + gettext_noop("Shows whether the instance is in recovery mode."), + NULL, + GUC_REPORT | GUC_NO_SHOW_ALL | GUC_NO_RESET_ALL | GUC_NOT_IN_SAMPLE | GUC_DISALLOW_IN_FILE + }, + &in_recovery, + false, + NULL, NULL, NULL + }, + { {"allow_system_table_mods", PGC_POSTMASTER, DEVELOPER_OPTIONS, gettext_noop("Allows modifications of the structure of system tables."), diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index da8b672096..86f0c13134 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -113,6 +113,7 @@ extern void CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conf extern int CountUserBackends(Oid roleid); extern bool CountOtherDBBackends(Oid databaseId, int *nbackends, int *nprepared); +extern void SendSignalToAllBackends(ProcSignalReason reason); extern void XidCacheRemoveRunningXids(TransactionId xid, int nxids, const TransactionId *xids, diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h index 05b186a05c..9cf9560b06 100644 --- a/src/include/storage/procsignal.h +++ b/src/include/storage/procsignal.h @@ -42,6 +42,8 @@ typedef enum PROCSIG_RECOVERY_CONFLICT_BUFFERPIN, PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK, + PROCSIG_RECOVERY_EXIT, /* recovery exit interrupt */ + NUM_PROCSIGNALS /* Must be last! */ } ProcSignalReason; diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h index a3f8f82ff3..2c73f0c0a8 100644 --- a/src/include/storage/standby.h +++ b/src/include/storage/standby.h @@ -26,6 +26,7 @@ extern int max_standby_streaming_delay; extern void InitRecoveryTransactionEnvironment(void); extern void ShutdownRecoveryTransactionEnvironment(void); +extern void SendRecoveryExitSignal(void); extern void ResolveRecoveryConflictWithSnapshot(TransactionId latestRemovedXid, RelFileNode node); diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h index 8dcfb40728..7b7bfe0209 100644 --- a/src/include/tcop/tcopprot.h +++ b/src/include/tcop/tcopprot.h @@ -71,6 +71,8 @@ extern void StatementCancelHandler(SIGNAL_ARGS); extern void FloatExceptionHandler(SIGNAL_ARGS) pg_attribute_noreturn(); extern void RecoveryConflictInterrupt(ProcSignalReason reason); /* called from SIGUSR1 * handler */ +/* recovery exit interrupt handling function */ +extern void HandleRecoveryExitInterrupt(void); extern void ProcessClientReadInterrupt(bool blocked); extern void ProcessClientWriteInterrupt(bool blocked); diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 94a537927a..6f13d099e2 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -2168,6 +2168,49 @@ reject_checked_write_connection(PGconn *conn) conn->try_next_host = true; } +static void +reject_checked_recovery_connection(PGconn *conn) +{ + /* Not a requested type; fail this connection. */ + const char *displayed_host; + const char *displayed_port; + + /* Append error report to conn->errorMessage. */ + if (conn->connhost[conn->whichhost].type == CHT_HOST_ADDRESS) + displayed_host = conn->connhost[conn->whichhost].hostaddr; + else + displayed_host = conn->connhost[conn->whichhost].host; + displayed_port = conn->connhost[conn->whichhost].port; + if (displayed_port == NULL || displayed_port[0] == '\0') + displayed_port = DEF_PGPORT_STR; + + if (conn->requested_session_type == SESSION_TYPE_PRIMARY) + appendPQExpBuffer(&conn->errorMessage, + libpq_gettext("server is in recovery mode " + "\"%s:%s\"\n"), + displayed_host, displayed_port); + else + appendPQExpBuffer(&conn->errorMessage, + libpq_gettext("server is not in recovery mode " + "\"%s:%s\"\n"), + displayed_host, displayed_port); + + /* Close connection politely. */ + conn->status = CONNECTION_OK; + sendTerminateConn(conn); + + /* Record primary host index */ + if (conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY && + conn->read_write_or_primary_host_index == -1) + conn->read_write_or_primary_host_index = conn->whichhost; + + /* + * Try next host if any, but we don't want to consider additional + * addresses for this host. + */ + conn->try_next_host = true; +} + /* ---------------- * PQconnectPoll * @@ -3621,27 +3664,52 @@ keep_going: /* We will come back to here until there is conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY || conn->requested_session_type == SESSION_TYPE_STANDBY))) { - /* - * Save existing error messages across the PQsendQuery - * attempt. This is necessary because PQsendQuery is - * going to reset conn->errorMessage, so we would lose - * error messages related to previous hosts we have tried - * and failed to connect to. - */ - if (!saveErrorMessage(conn, &savedMessage)) - goto error_return; - conn->status = CONNECTION_OK; - if (!PQsendQuery(conn, "SELECT pg_is_in_recovery()")) + if (conn->sversion < 120000) { + /* + * Save existing error messages across the PQsendQuery + * attempt. This is necessary because PQsendQuery is + * going to reset conn->errorMessage, so we would lose + * error messages related to previous hosts we have + * tried and failed to connect to. + */ + if (!saveErrorMessage(conn, &savedMessage)) + goto error_return; + + conn->status = CONNECTION_OK; + if (!PQsendQuery(conn, "SELECT pg_is_in_recovery()")) + { + restoreErrorMessage(conn, &savedMessage); + goto error_return; + } + + conn->status = CONNECTION_CHECK_RECOVERY; + restoreErrorMessage(conn, &savedMessage); - goto error_return; + return PGRES_POLLING_READING; } + else if ((conn->in_recovery && + conn->requested_session_type == SESSION_TYPE_PRIMARY) || + (!conn->in_recovery && + (conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY || + conn->requested_session_type == SESSION_TYPE_STANDBY))) + { + /* + * The following scenario is possible only for the + * prefer-standby mode for the next pass of the list + * of connections as it couldn't find any servers that + * are in recovery. + */ + if (conn->read_write_or_primary_host_index == -2) + goto consume_checked_target_connection; - conn->status = CONNECTION_CHECK_RECOVERY; + reject_checked_recovery_connection(conn); + goto keep_going; + } - restoreErrorMessage(conn, &savedMessage); - return PGRES_POLLING_READING; + /* obtained the requested type, consume it */ + goto consume_checked_target_connection; } /* @@ -3890,40 +3958,7 @@ keep_going: /* We will come back to here until there is PQclear(res); restoreErrorMessage(conn, &savedMessage); - /* Append error report to conn->errorMessage. */ - if (conn->connhost[conn->whichhost].type == CHT_HOST_ADDRESS) - displayed_host = conn->connhost[conn->whichhost].hostaddr; - else - displayed_host = conn->connhost[conn->whichhost].host; - displayed_port = conn->connhost[conn->whichhost].port; - if (displayed_port == NULL || displayed_port[0] == '\0') - displayed_port = DEF_PGPORT_STR; - - if (conn->requested_session_type == SESSION_TYPE_PRIMARY) - appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("server is in recovery mode " - "\"%s:%s\"\n"), - displayed_host, displayed_port); - else - appendPQExpBuffer(&conn->errorMessage, - libpq_gettext("server is not in recovery mode " - "\"%s:%s\"\n"), - displayed_host, displayed_port); - - /* Close connection politely. */ - conn->status = CONNECTION_OK; - sendTerminateConn(conn); - - /* Record primary host index */ - if (conn->requested_session_type == SESSION_TYPE_PREFER_STANDBY && - conn->read_write_or_primary_host_index == -1) - conn->read_write_or_primary_host_index = conn->whichhost; - - /* - * Try next host if any, but we don't want to consider - * additional addresses for this host. - */ - conn->try_next_host = true; + reject_checked_recovery_connection(conn); goto keep_going; } diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 701c4a66fe..7cffed57cd 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -1117,6 +1117,10 @@ pqSaveParameterStatus(PGconn *conn, const char *name, const char *value) { conn->transaction_read_only = (strcmp(value, "on") == 0); } + else if (strcmp(name, "in_recovery") == 0) + { + conn->in_recovery = (strcmp(value, "on") == 0); + } } diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 740033e116..7b36ddb804 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -443,6 +443,7 @@ struct pg_conn int client_encoding; /* encoding id */ bool std_strings; /* standard_conforming_strings */ bool transaction_read_only; /* transaction_read_only */ + bool in_recovery; /* in_recovery */ PGVerbosity verbosity; /* error/notice message verbosity */ PGContextVisibility show_context; /* whether to show CONTEXT field */ PGlobjfuncs *lobjfuncs; /* private state for large-object access fns */ -- 2.20.1.windows.1