public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'Fujii Masao' <[email protected]>
To: 'Kyotaro Horiguchi' <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: [Proposal] Add foreign-server health checks infrastructure
Date: Thu, 24 Feb 2022 02:34:55 +0000
Message-ID: <TYAPR01MB5866FC683843ED8BD09505FEF53D9@TYAPR01MB5866.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<[email protected]>
<TYAPR01MB58661B088B6066282824ADD9F5369@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
<TYAPR01MB58664E79C4728A6FFEA46F94F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
<TYAPR01MB5866FF28C802916242F1B492F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<TYAPR01MB5866C6278B6386672B35F624F53B9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<[email protected]>
Dear Fujii-san,
Thank you for your quick reviewing! I attached new version.
I found previous patches have wrong name. Sorry.
> The connection check timer is re-scheduled repeatedly even while the backend is
> in idle state or is running a local transaction that doesn't access to any foreign
> servers. I'm not sure if it's really worth checking the connections even in those
> states. Even without the periodic connection checks, if the connections are closed
> in those states, subsequent GetConnection() will detect that closed connection
> and re-establish the connection when starting remote transaction. Thought?
Indeed. We can now control the timer in fdw layer, so disable_timeout() was added
at the bottom of pgfdw_xact_callback().
> When a closed connection is detected in idle-in-transaction state and SIGINT is
> raised, nothing happens because there is no query running to be canceled by
> SIGINT. Also in this case the connection check timer gets disabled. So we can still
> execute queries that don't access to foreign servers, in the same transaction, and
> then the transaction commit fails. Is this expected behavior?
It's not happy, but I'm not sure about a good solution. I made a timer reschedule
if connection lost had detected. But if queries in the transaction are quite short,
catching SIGINT may be fail.
> When I shutdowned the foreign server while the local backend is in
> idle-in-transaction state, the connection check timer was triggered and detected
> the closed connection. Then when I executed COMMIT command, I got the
> following WARNING message. Is this a bug?
>
> WARNING: leaked hash_seq_search scan for hash table 0x7fd2ca878f20
Fixed. It is caused because hash_seq_term() was not called when checker detects
a connection lost.
Best Regards,
Hayato Kuroda
FUJITSU LIMITED
Attachments:
[application/octet-stream] v13_0001_expose_cancel_message.patch (1.5K, ../TYAPR01MB5866FC683843ED8BD09505FEF53D9@TYAPR01MB5866.jpnprd01.prod.outlook.com/2-v13_0001_expose_cancel_message.patch)
download | inline diff:
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 3c7d08209f..206f7e1d59 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -106,6 +106,8 @@ int PostAuthDelay = 0;
/* Time between checks that the client is still connected. */
int client_connection_check_interval = 0;
+char *QeuryCancelMessage = NULL;
+
/* ----------------
* private typedefs etc
* ----------------
@@ -3327,6 +3329,8 @@ ProcessInterrupts(void)
LockErrorCleanup();
ereport(ERROR,
(errcode(ERRCODE_QUERY_CANCELED),
+ QeuryCancelMessage ?
+ errmsg("%s", QeuryCancelMessage) :
errmsg("canceling statement due to user request")));
}
}
@@ -4248,6 +4252,9 @@ PostgresMain(const char *dbname, const char *username)
/* Report the error to the client and/or server log */
EmitErrorReport();
+ /* Make sure QeuryCancelMessage is reset. */
+ QeuryCancelMessage = NULL;
+
/*
* Make sure debug_query_string gets reset before we possibly clobber
* the storage it points at.
diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h
index 15a11bc3ff..83f894cbd0 100644
--- a/src/include/tcop/tcopprot.h
+++ b/src/include/tcop/tcopprot.h
@@ -30,6 +30,7 @@ extern PGDLLIMPORT const char *debug_query_string;
extern int max_stack_depth;
extern int PostAuthDelay;
extern int client_connection_check_interval;
+extern PGDLLIMPORT char* QeuryCancelMessage;
/* GUC-configurable parameters */
[application/octet-stream] v13_0002_add_health_check.patch (7.1K, ../TYAPR01MB5866FC683843ED8BD09505FEF53D9@TYAPR01MB5866.jpnprd01.prod.outlook.com/3-v13_0002_add_health_check.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index f753c6e232..f691b482a1 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -23,12 +23,14 @@
#include "postgres_fdw.h"
#include "storage/fd.h"
#include "storage/latch.h"
+#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/datetime.h"
#include "utils/hsearch.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/syscache.h"
+#include "utils/timeout.h"
/*
* Connection cache hash table entry
@@ -109,6 +111,10 @@ static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql,
bool toplevel);
static bool UserMappingPasswordRequired(UserMapping *user);
static bool disconnect_cached_connections(Oid serverid);
+static void pgfdw_connection_check(void);
+static bool pgfdw_connection_check_internal(PGconn *conn);
+static TimeoutId pgfdw_health_check_timeout = MAX_TIMEOUTS;
+int pgfdw_health_check_interval;
/*
* Get a PGconn which can be used to execute queries on the remote PostgreSQL
@@ -153,6 +159,9 @@ GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
pgfdw_inval_callback, (Datum) 0);
CacheRegisterSyscacheCallback(USERMAPPINGOID,
pgfdw_inval_callback, (Datum) 0);
+
+ /* Register a timeout for checking remote servers */
+ pgfdw_health_check_timeout = RegisterTimeout(USER_TIMEOUT, pgfdw_connection_check);
}
/* Set flag that we did GetConnection during the current transaction */
@@ -276,6 +285,12 @@ GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
if (state)
*state = &entry->state;
+ /* Fire timeout if needed */
+ if (pgfdw_health_check_interval > 0 &&
+ !get_timeout_active(pgfdw_health_check_timeout))
+ enable_timeout_after(pgfdw_health_check_timeout,
+ pgfdw_health_check_interval);
+
return entry->conn;
}
@@ -1007,6 +1022,9 @@ pgfdw_xact_callback(XactEvent event, void *arg)
*/
xact_got_connection = false;
+ /* stop timer because checking is no more needed. */
+ disable_timeout(pgfdw_health_check_timeout, false);
+
/* Also reset cursor numbering for next transaction */
cursor_number = 0;
}
@@ -1702,3 +1720,139 @@ disconnect_cached_connections(Oid serverid)
return result;
}
+
+/*
+ * Signal handler for checking remote servers.
+ *
+ * This function searches the hash table from the beginning
+ * and performs a health-check on each entry.
+ *
+ * Raise SIGINT if someone might be down, otherwise do nothing.
+ */
+void
+pgfdw_connection_check(void)
+{
+ HASH_SEQ_STATUS scan;
+ ConnCacheEntry *entry;
+
+ Assert(ConnectionHash);
+
+ /*
+ * checking will be done by waiting WL_SOCKET_CLOSED event,
+ * so exit immediately if it cannot be used in this system.
+ */
+ if (!WaitEventSetCanReportClosed())
+ return;
+
+ /* Is there any cancel messages? If so, raise again and re-schedule */
+ if (QeuryCancelMessage != NULL)
+ {
+ raise(SIGINT);
+ if (pgfdw_health_check_interval > 0)
+ enable_timeout_after(pgfdw_health_check_timeout,
+ pgfdw_health_check_interval);
+ return;
+ }
+
+ hash_seq_init(&scan, ConnectionHash);
+ while ((entry = (ConnCacheEntry *) hash_seq_search(&scan)))
+ {
+ if (entry->conn == NULL || entry->xact_depth == 0)
+ continue;
+ if (!pgfdw_connection_check_internal(entry->conn))
+ {
+ /*
+ * Foreign server might be down, so raise SIGINT.
+ * Note that error message is passed to QeuryCancelMessage
+ * for reporting error in ProcessInterrupts().
+ */
+ char msg[31 + MAXDATELEN];
+ MemoryContext old;
+ ForeignServer *server;
+
+ /*
+ * Switch to CurTransactionContext in order to
+ * make sure that the lifetime of palloc'd is transaction.
+ */
+ old = MemoryContextSwitchTo(CurTransactionContext);
+ server = GetForeignServer(entry->serverid);
+ snprintf(msg, sizeof(msg), "Foreign Server %s might be down.", server->servername);
+ QeuryCancelMessage = pstrdup(msg);
+ MemoryContextSwitchTo(old);
+
+ disconnect_pg_server(entry);
+ hash_seq_term(&scan);
+
+ raise(SIGINT);
+ break;
+ }
+ }
+
+ /* re-schedule timer if needed. */
+ if (pgfdw_health_check_interval > 0)
+ enable_timeout_after(pgfdw_health_check_timeout,
+ pgfdw_health_check_interval);
+
+ return;
+}
+
+/*
+ * helper function for pgfdw_connection_check
+ */
+static bool
+pgfdw_connection_check_internal(PGconn *conn)
+{
+ WaitEventSet *eventset;
+ WaitEvent events;
+
+ Assert(WaitEventSetCanReportClosed());
+
+ eventset = CreateWaitEventSet(CurrentMemoryContext, 1);
+ AddWaitEventToSet(eventset, WL_SOCKET_CLOSED, PQsocket(conn), NULL, NULL);
+
+ WaitEventSetWait(eventset, 0, &events, 1, 0);
+
+ if (events.events & WL_SOCKET_CLOSED)
+ {
+ FreeWaitEventSet(eventset);
+ return false;
+ }
+ FreeWaitEventSet(eventset);
+
+ return true;
+}
+
+bool
+check_pgfdw_health_check_interval(int *newval, void **extra, GucSource source)
+{
+ if (!WaitEventSetCanReportClosed() && *newval != 0)
+ {
+ GUC_check_errdetail("pgfdw_health_check_interval must be set to 0 on this platform");
+ return false;
+ }
+ return true;
+}
+
+void
+assign_pgfdw_health_check_interval(int newval, void *extra)
+{
+ /* Quick return if timeout is not registered yet. */
+ if (pgfdw_health_check_timeout == MAX_TIMEOUTS)
+ return;
+
+ if (get_timeout_active(pgfdw_health_check_timeout))
+ {
+ if (newval == 0)
+ disable_timeout(pgfdw_health_check_timeout, false);
+
+ /*
+ * we don't have to do anything because
+ * new value will be used in pgfdw_connection_check().
+ */
+ return;
+ }
+
+ /* Start timeout if wants to */
+ if (newval > 0)
+ enable_timeout_after(pgfdw_health_check_timeout, newval);
+}
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index 2c6b2894b9..75a910b0ff 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -538,5 +538,18 @@ _PG_init(void)
NULL,
NULL);
+ DefineCustomIntVariable("postgres_fdw.health_check_interval",
+ "Sets the time interval between checks of remote servers.",
+ NULL,
+ &pgfdw_health_check_interval,
+ 0,
+ 0,
+ INT_MAX,
+ PGC_USERSET,
+ GUC_UNIT_MS,
+ check_pgfdw_health_check_interval,
+ assign_pgfdw_health_check_interval,
+ NULL);
+
MarkGUCPrefixReserved("postgres_fdw");
}
diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h
index 8ae79e97e4..c129af5082 100644
--- a/contrib/postgres_fdw/postgres_fdw.h
+++ b/contrib/postgres_fdw/postgres_fdw.h
@@ -18,6 +18,7 @@
#include "libpq-fe.h"
#include "nodes/execnodes.h"
#include "nodes/pathnodes.h"
+#include "utils/guc.h"
#include "utils/relcache.h"
/*
@@ -151,6 +152,10 @@ extern PGresult *pgfdw_exec_query(PGconn *conn, const char *query,
PgFdwConnState *state);
extern void pgfdw_report_error(int elevel, PGresult *res, PGconn *conn,
bool clear, const char *sql);
+extern bool check_pgfdw_health_check_interval(int *newval, void **extra,
+ GucSource source);
+extern void assign_pgfdw_health_check_interval(int newval, void *extra);
+extern int pgfdw_health_check_interval;
/* in option.c */
extern int ExtractConnectionOptions(List *defelems,
[application/octet-stream] v13_0003_add_doc.patch (1.4K, ../TYAPR01MB5866FC683843ED8BD09505FEF53D9@TYAPR01MB5866.jpnprd01.prod.outlook.com/4-v13_0003_add_doc.patch)
download | inline diff:
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index dc57fe4b0d..21532e19d9 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -1028,6 +1028,32 @@ postgres=# SELECT postgres_fdw_disconnect_all();
</listitem>
</varlistentry>
+ <varlistentry id="guc-pgfdw-health-check-interval" xreflabel="postgres_fdw.health_check_interval">
+ <term>
+ <varname>postgres_fdw.health_check_interval</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>postgres_fdw.health_check_interval</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Sets the time interval between optional checks that remote servers
+ are still alive. When losing a remote connection is detected,
+ the running transaction is aborted. This feature is performed
+ by polling the socket.
+ </para>
+ <para>
+ this option relies on kernel events exposed by Linux, macOS,
+ illumos and the BSD family of operating systems, and is not currently available
+ on other systems.
+ </para>
+ <para>
+ If the value is specified without units, it is taken as milliseconds.
+ The default value is <literal>0</literal>, which disables connection
+ checks.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
</sect2>
[application/x-zip-compressed] v13_0004_add_test.zip (855B, ../TYAPR01MB5866FC683843ED8BD09505FEF53D9@TYAPR01MB5866.jpnprd01.prod.outlook.com/5-v13_0004_add_test.zip)
download
view thread (30+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: [Proposal] Add foreign-server health checks infrastructure
In-Reply-To: <TYAPR01MB5866FC683843ED8BD09505FEF53D9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox