diff --git a/src/backend/commands/copyfromparse.c b/src/backend/commands/copyfromparse.c index aac10165ec..0c4d61a34c 100644 --- a/src/backend/commands/copyfromparse.c +++ b/src/backend/commands/copyfromparse.c @@ -269,6 +269,7 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread) readmessage: HOLD_CANCEL_INTERRUPTS(); + HOLD_CHECKING_REMOTE_SERVERS_INTERRUPTS(); pq_startmsgread(); mtype = pq_getbyte(); if (mtype == EOF) @@ -300,6 +301,7 @@ CopyGetData(CopyFromState cstate, void *databuf, int minread, int maxread) ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("unexpected EOF on client connection with an open transaction"))); + RESUME_CHECKING_REMOTE_SERVERS_INTERRUPTS(); RESUME_CANCEL_INTERRUPTS(); /* ... and process it */ switch (mtype) diff --git a/src/backend/foreign/foreign.c b/src/backend/foreign/foreign.c index e07cc57431..d6a6d3d215 100644 --- a/src/backend/foreign/foreign.c +++ b/src/backend/foreign/foreign.c @@ -14,6 +14,7 @@ #include "access/htup_details.h" #include "access/reloptions.h" +#include "access/xact.h" #include "catalog/pg_foreign_data_wrapper.h" #include "catalog/pg_foreign_server.h" #include "catalog/pg_foreign_table.h" @@ -26,7 +27,13 @@ #include "utils/memutils.h" #include "utils/rel.h" #include "utils/syscache.h" +#include "utils/timeout.h" +/* for checking remote servers */ +int remote_servers_connection_check_interval = 0; +static CheckingRemoteServersCallbackItem *fdw_callbacks = NULL; +static bool isRegistered = false; +static void UnregisterCheckingCallbacks(XactEvent event, void *arg); /* * GetForeignDataWrapper - look up the foreign-data wrapper by OID. @@ -836,3 +843,144 @@ GetExistingLocalJoinPath(RelOptInfo *joinrel) } return NULL; } + +/* + * Register callbacks for checking remote servers. + * + * This function is intended for use by FDW extensions. + * The checking timeout will be fired after registering the first callback. + */ +void +RegisterCheckingRemoteServersCallback(CheckingRemoteServersCallback callback, + void *arg) +{ + CheckingRemoteServersCallbackItem *item; + /* should we start checking timeout? */ + bool shouldStart = !HaveCheckingRemoteServersCallbacks(); + + item = (CheckingRemoteServersCallbackItem *) + MemoryContextAlloc(TopMemoryContext, + sizeof(CheckingRemoteServersCallbackItem)); + item->callback = callback; + item->arg = arg; + item->next = fdw_callbacks; + fdw_callbacks = item; + + if (shouldStart && remote_servers_connection_check_interval > 0) + enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT, + remote_servers_connection_check_interval); + + if (!isRegistered) + { + RegisterXactCallback(UnregisterCheckingCallbacks, NULL); + isRegistered = true; + } +} + +static void +UnregisterCheckingCallbacks(XactEvent event, void *arg) +{ + if (event != XACT_EVENT_PRE_COMMIT && + event != XACT_EVENT_PRE_PREPARE) + UnregisterAllCheckingRemoteServersCallback(); +} + +/* + * Unregister the specified callback. + */ +void +UnregisterCheckingRemoteServersCallback(CheckingRemoteServersCallback callback, + void *arg) +{ + CheckingRemoteServersCallbackItem *item; + CheckingRemoteServersCallbackItem *prev; + + prev = NULL; + for (item = fdw_callbacks; item; prev = item, item = item->next) + { + if (item->callback == callback && item->arg == arg) + { + if (prev) + prev->next = item->next; + else + fdw_callbacks = item->next; + pfree(item); + break; + } + } + if (!HaveCheckingRemoteServersCallbacks() && + get_timeout_active(CHECKING_REMOTE_SERVERS_TIMEOUT)) + disable_timeout(CHECKING_REMOTE_SERVERS_TIMEOUT, false); +} + +/* + * Unregister all callbacks. + * + * This function is called when commit/abort any transactions. + */ +void +UnregisterAllCheckingRemoteServersCallback(void) +{ + CheckingRemoteServersCallbackItem *item; + item = fdw_callbacks; + while (item) + { + CheckingRemoteServersCallbackItem *next; + next = item->next; + pfree(item); + item = next; + } + fdw_callbacks = NULL; + if (get_timeout_active(CHECKING_REMOTE_SERVERS_TIMEOUT)) + disable_timeout(CHECKING_REMOTE_SERVERS_TIMEOUT, false); +} + +/* + * Call callbacks for checking remote servers. + * + * Note that this function will not return anything. + * Callback functions must throw ereport(ERROR) + * if disconnection has been detected. + */ +void +CallCheckingRemoteServersCallbacks(void) +{ + CheckingRemoteServersCallbackItem *item; + for (item = fdw_callbacks; item; item = item->next) + item->callback(item->arg); +} + +/* + * Check whether any callbacks has been registered. + */ +bool +HaveCheckingRemoteServersCallbacks(void) +{ + return fdw_callbacks != NULL; +} + +void +assign_remote_servers_connection_check_interval(int newval, + void *extra) +{ + /* Quick return if we don't have any callbacks */ + if (!HaveCheckingRemoteServersCallbacks()) + return; + + if (get_timeout_active(CHECKING_REMOTE_SERVERS_TIMEOUT)) + { + /* disable timeout if zero */ + if (newval == 0) + disable_timeout(CHECKING_REMOTE_SERVERS_TIMEOUT, false); + + /* + * we don't have to do anything because + * new value will be used in ProcessInterrupts(). + */ + return; + } + + if (newval > 0) + enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT, + newval); +} diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 82de01cdc6..de0ed3f7b6 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -44,6 +44,7 @@ #include "commands/prepare.h" #include "common/pg_prng.h" #include "executor/spi.h" +#include "foreign/foreign.h" #include "jit/jit.h" #include "libpq/libpq.h" #include "libpq/pqformat.h" @@ -350,6 +351,7 @@ SocketBackend(StringInfo inBuf) * Get message type code from the frontend. */ HOLD_CANCEL_INTERRUPTS(); + HOLD_CHECKING_REMOTE_SERVERS_INTERRUPTS(); pq_startmsgread(); qtype = pq_getbyte(); @@ -456,6 +458,7 @@ SocketBackend(StringInfo inBuf) */ if (pq_getmessage(inBuf, maxmsglen)) return EOF; /* suitable message already logged */ + RESUME_CHECKING_REMOTE_SERVERS_INTERRUPTS(); RESUME_CANCEL_INTERRUPTS(); return qtype; @@ -2709,6 +2712,13 @@ start_xact_command(void) !get_timeout_active(CLIENT_CONNECTION_CHECK_TIMEOUT)) enable_timeout_after(CLIENT_CONNECTION_CHECK_TIMEOUT, client_connection_check_interval); + if (remote_servers_connection_check_interval > 0 && + IsUnderPostmaster && + MyProcPort && + HaveCheckingRemoteServersCallbacks() && + !get_timeout_active(CHECKING_REMOTE_SERVERS_TIMEOUT)) + enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT, + remote_servers_connection_check_interval); } static void @@ -3214,6 +3224,27 @@ ProcessInterrupts(void) } } + if (CheckingRemoteServersTimeoutPending) + { + if (CheckingRemoteServersHoldoffCount != 0) + { + /* + * Skip checking foreign servers while reading messages. + */ + InterruptPending = true; + } + else + { + CheckingRemoteServersTimeoutPending = false; + + CallCheckingRemoteServersCallbacks(); + + if (remote_servers_connection_check_interval > 0) + enable_timeout_after(CHECKING_REMOTE_SERVERS_TIMEOUT, + remote_servers_connection_check_interval); + } + } + if (ClientConnectionLost) { QueryCancelPending = false; /* lost connection trumps QueryCancel */ diff --git a/src/backend/utils/init/globals.c b/src/backend/utils/init/globals.c index 381d9e548d..15b0c2727b 100644 --- a/src/backend/utils/init/globals.c +++ b/src/backend/utils/init/globals.c @@ -36,9 +36,11 @@ volatile sig_atomic_t IdleInTransactionSessionTimeoutPending = false; volatile sig_atomic_t IdleSessionTimeoutPending = false; volatile sig_atomic_t ProcSignalBarrierPending = false; volatile sig_atomic_t LogMemoryContextPending = false; +volatile sig_atomic_t CheckingRemoteServersTimeoutPending = false; volatile uint32 InterruptHoldoffCount = 0; volatile uint32 QueryCancelHoldoffCount = 0; volatile uint32 CritSectionCount = 0; +volatile uint32 CheckingRemoteServersHoldoffCount = 0; int MyProcPid; pg_time_t MyStartTime; diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c index 7292e51f7d..3f936be326 100644 --- a/src/backend/utils/init/postinit.c +++ b/src/backend/utils/init/postinit.c @@ -73,6 +73,7 @@ static void LockTimeoutHandler(void); static void IdleInTransactionSessionTimeoutHandler(void); static void IdleSessionTimeoutHandler(void); static void ClientCheckTimeoutHandler(void); +static void CheckingRemoteServersTimeoutHandler(void); static bool ThereIsAtLeastOneRole(void); static void process_startup_options(Port *port, bool am_superuser); static void process_settings(Oid databaseid, Oid roleid); @@ -621,6 +622,7 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username, IdleInTransactionSessionTimeoutHandler); RegisterTimeout(IDLE_SESSION_TIMEOUT, IdleSessionTimeoutHandler); RegisterTimeout(CLIENT_CONNECTION_CHECK_TIMEOUT, ClientCheckTimeoutHandler); + RegisterTimeout(CHECKING_REMOTE_SERVERS_TIMEOUT, CheckingRemoteServersTimeoutHandler); } /* @@ -1239,6 +1241,14 @@ ClientCheckTimeoutHandler(void) SetLatch(MyLatch); } +static void +CheckingRemoteServersTimeoutHandler(void) +{ + CheckingRemoteServersTimeoutPending = true; + InterruptPending = true; + SetLatch(MyLatch); +} + /* * Returns true if at least one role is defined in this database cluster. */ diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index f736e8d872..763f8198e5 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -51,6 +51,7 @@ #include "commands/vacuum.h" #include "commands/variable.h" #include "common/string.h" +#include "foreign/foreign.h" #include "funcapi.h" #include "jit/jit.h" #include "libpq/auth.h" @@ -105,6 +106,7 @@ #include "utils/queryjumble.h" #include "utils/rls.h" #include "utils/snapmgr.h" +#include "utils/timeout.h" #include "utils/tzparser.h" #include "utils/inval.h" #include "utils/varlena.h" @@ -3590,6 +3592,18 @@ static struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + + { + {"remote_servers_connection_check_interval", PGC_USERSET, CONN_AUTH_SETTINGS, + gettext_noop("Sets the time interval between checks for disconnection of remote servers."), + NULL, + GUC_UNIT_MS + }, + &remote_servers_connection_check_interval, + 0, 0, INT_MAX, + NULL, assign_remote_servers_connection_check_interval, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index a1acd46b61..cad3c72336 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -90,6 +90,10 @@ # disconnection while running queries; # 0 for never +#remote_servers_connection_check_interval = 0 # time between time between checks for + # foreign server disconnection; + # 0 for never + # - Authentication - #authentication_timeout = 1min # 1s-600s diff --git a/src/include/foreign/foreign.h b/src/include/foreign/foreign.h index 8169eb76b1..a1e01d91e4 100644 --- a/src/include/foreign/foreign.h +++ b/src/include/foreign/foreign.h @@ -81,4 +81,26 @@ extern List *GetForeignColumnOptions(Oid relid, AttrNumber attnum); extern Oid get_foreign_data_wrapper_oid(const char *fdwname, bool missing_ok); extern Oid get_foreign_server_oid(const char *servername, bool missing_ok); + +/* functions and variables for fdw checking. */ +typedef void (*CheckingRemoteServersCallback) (void *arg); +typedef struct CheckingRemoteServersCallbackItem +{ + struct CheckingRemoteServersCallbackItem *next; + CheckingRemoteServersCallback callback; + void *arg; +} CheckingRemoteServersCallbackItem; + +extern void RegisterCheckingRemoteServersCallback(CheckingRemoteServersCallback callback, + void *arg); +extern void UnregisterCheckingRemoteServersCallback(CheckingRemoteServersCallback callback, + void *arg); +extern void UnregisterAllCheckingRemoteServersCallback(void); +extern void CallCheckingRemoteServersCallbacks(void); +extern bool HaveCheckingRemoteServersCallbacks(void); + +extern int remote_servers_connection_check_interval; +extern void assign_remote_servers_connection_check_interval(int newval, + void *extra); + #endif /* FOREIGN_H */ diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 90a3016065..377fc68aaa 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -98,10 +98,13 @@ extern PGDLLIMPORT volatile sig_atomic_t LogMemoryContextPending; extern PGDLLIMPORT volatile sig_atomic_t CheckClientConnectionPending; extern PGDLLIMPORT volatile sig_atomic_t ClientConnectionLost; +extern PGDLLIMPORT volatile sig_atomic_t CheckingRemoteServersTimeoutPending; + /* these are marked volatile because they are examined by signal handlers: */ extern PGDLLIMPORT volatile uint32 InterruptHoldoffCount; extern PGDLLIMPORT volatile uint32 QueryCancelHoldoffCount; extern PGDLLIMPORT volatile uint32 CritSectionCount; +extern PGDLLIMPORT volatile uint32 CheckingRemoteServersHoldoffCount; /* in tcop/postgres.c */ extern void ProcessInterrupts(void); @@ -126,7 +129,7 @@ do { \ /* Is ProcessInterrupts() guaranteed to clear InterruptPending? */ #define INTERRUPTS_CAN_BE_PROCESSED() \ (InterruptHoldoffCount == 0 && CritSectionCount == 0 && \ - QueryCancelHoldoffCount == 0) + QueryCancelHoldoffCount == 0 && CheckingRemoteServersHoldoffCount == 0) #define HOLD_INTERRUPTS() (InterruptHoldoffCount++) @@ -152,6 +155,13 @@ do { \ CritSectionCount--; \ } while(0) +#define HOLD_CHECKING_REMOTE_SERVERS_INTERRUPTS() (CheckingRemoteServersHoldoffCount++) + +#define RESUME_CHECKING_REMOTE_SERVERS_INTERRUPTS() \ +do { \ + Assert(CheckingRemoteServersHoldoffCount > 0); \ + CheckingRemoteServersHoldoffCount--; \ +} while(0) /***************************************************************************** * globals.h -- * diff --git a/src/include/utils/timeout.h b/src/include/utils/timeout.h index 2cbc5de4d9..ceb6b1c12c 100644 --- a/src/include/utils/timeout.h +++ b/src/include/utils/timeout.h @@ -34,6 +34,7 @@ typedef enum TimeoutId IDLE_SESSION_TIMEOUT, CLIENT_CONNECTION_CHECK_TIMEOUT, STARTUP_PROGRESS_TIMEOUT, + CHECKING_REMOTE_SERVERS_TIMEOUT, /* First user-definable timeout reason */ USER_TIMEOUT, /* Maximum number of timeout reasons */