Disable timeouts for multimaster

To avoid disconnection between multimaster components
the following timeouts were disabled for the multimaster:
- statement_timeout;
- lock_timeout;
- idle_in_transaction_session_timeout;
- idle_session_timeout.

Author: Mikhail Rutman

Makefile

@@ -4,7 +4,7 @@ OBJS = src/multimaster.o src/dmq.o src/commit.o src/bytebuf.o src/bgwpool.o \
 src/pglogical_output.o src/pglogical_proto.o src/pglogical_receiver.o \
 src/pglogical_apply.o src/pglogical_hooks.o src/pglogical_config.o \
 src/pglogical_relid_map.o src/ddd.o src/bkb.o src/spill.o src/state.o \
-src/resolver.o src/ddl.o src/syncpoint.o src/global_tx.o
+src/resolver.o src/ddl.o src/syncpoint.o src/global_tx.o src/mtm_utils.o
 MODULE_big = multimaster
 
 ifndef USE_PGXS # hmm, user didn't requested to use pgxs

src/bgwpool.c

@@ -31,6 +31,7 @@
 #include "multimaster.h"
 #include "state.h"
 #include "logger.h"
+#include "mtm_utils.h"
 
 /*
  * Store the size of tx body, position of it in the tx list and transaction
@@ -324,6 +325,7 @@ BgwPoolMainLoop(BgwPool *poolDesc)
 void
 BgwPoolDynamicWorkerMainLoop(Datum arg)
 {
+	MtmDisableTimeouts();
 	BgwPoolMainLoop((BgwPool *) DatumGetPointer(arg));
 }
 

src/ddl.c

@@ -666,7 +666,6 @@ MtmProcessUtility(PlannedStmt *pstmt, const char *queryString,
 				  QueryEnvironment *queryEnv, DestReceiver *dest,
 				  QueryCompletion *qc)
 {
-
 	/*
 	 * Quick exit if multimaster is not enabled.
 	 * XXX it's better to do MtmIsEnabled here, but this needs cache access

src/dmq.c

@@ -31,6 +31,7 @@
 #include "dmq.h"
 #include "logger.h"
 #include "compat.h"
+#include "mtm_utils.h"
 
 #include "access/transam.h"
 #include "libpq/libpq.h"
@@ -527,6 +528,8 @@ dmq_sender_main(Datum main_arg)
 	pqsignal(SIGTERM, die);
 	BackgroundWorkerUnblockSignals();
 
+	MtmDisableTimeouts();
+
 	memcpy(&heartbeat_send_timeout, MyBgworkerEntry->bgw_extra, sizeof(int));
 	memcpy(&connect_timeout, MyBgworkerEntry->bgw_extra + sizeof(int), sizeof(int));
 
@@ -796,7 +799,7 @@ dmq_sender_main(Datum main_arg)
 						int			pos = event.pos;
 
 						pqtime = dmq_now();
-						status = PQconnectPoll(conns[conn_id].pgconn);
+						status = MtmPQconnectPoll(conns[conn_id].pgconn);
 						mtm_log(DmqPqTiming, "[DMQ] [TIMING] pqp = %f ms", dmq_now() - pqtime);
 
 						mtm_log(DmqStateIntermediate,
@@ -1386,6 +1389,11 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 	double		last_message_at = dmq_now();
 	void		*extra = NULL;
 
+	/*
+	 * We do not call MtmDisbaleTimeouts() here because of connection to this
+	 * client is made by MtmPQconnectPoll() that sets all needed timeouts.
+	 */
+
 	sender_name = text_to_cstring(PG_GETARG_TEXT_PP(0));
 	recv_timeout = PG_GETARG_INT32(1);
 

src/include/mtm_utils.h

@@ -0,0 +1,24 @@
+/*-------------------------------------------------------------------------
+ *
+ * mtm_utils.h
+ *	Utility functions:
+ *	- disable global timeouts settings;
+ *	- libpq connect function wrappers.
+ *
+ *
+ * Copyright (c) 2022, Postgres Professional
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef MTM_UTILS_H
+#define MTM_UTILS_H
+
+#include "libpq/pqformat.h"
+#include "libpq-fe.h"
+
+extern void MtmDisableTimeouts(void);
+
+extern PostgresPollingStatusType MtmPQconnectPoll(PGconn *conn);
+extern PGconn* MtmPQconnectdb(const char *conninfo);
+
+#endif

src/mtm_utils.c

@@ -0,0 +1,133 @@
+/*----------------------------------------------------------------------------
+ *
+ * mtm_utils.c
+ *	  Utility functions
+ *
+ * Copyright (c) 2022, Postgres Professional
+ *
+ *----------------------------------------------------------------------------
+ */
+
+#include "logger.h"
+#include "mtm_utils.h"
+
+#include "utils/timeout.h"
+
+/*
+ * Disables timeouts on a client side:
+ * - statement_timeout;
+ * - lock_timeout;
+ * - idle_in_transaction_session_timeout;
+ * - idle_session_timeout.
+ *
+ * This timeouts, when set in the postgres config file, affect all process.
+ * The multimaster needs his sessions not to be interrupted, so we disable
+ * these timeouts. 
+ *
+ * This function raises an error on PQExec failed.
+ */
+static void
+disable_client_timeouts(PGconn *conn)
+{
+	PGresult   *res;
+
+	res = PQexec(conn, "SET statement_timeout = 0");
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		char	   *msg = pchomp(PQerrorMessage(conn));
+		mtm_log(ERROR, "failed to set statement_timeout: %s", msg);
+	}
+
+	res = PQexec(conn, "SET lock_timeout = 0");
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		char	   *msg = pchomp(PQerrorMessage(conn));
+		mtm_log(ERROR, "failed to set lock_timeout: %s", msg);
+	}
+
+	res = PQexec(conn, "SET idle_in_transaction_session_timeout = 0");
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		char	   *msg = pchomp(PQerrorMessage(conn));
+		mtm_log(ERROR, "failed to set idle_in_transaction_session_timeout: %s", msg);
+	}
+
+	res = PQexec(conn, "SET idle_session_timeout = 0");
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		char	   *msg = pchomp(PQerrorMessage(conn));
+		mtm_log(ERROR, "failed to set idle_session_timeout: %s", msg);
+	}
+}
+
+/*
+ * Disable timeouts for a current process
+ * - statement_timeout;
+ * - lock_timeout;
+ * - idle_in_transaction_session_timeout;
+ * - idle_session_timeout.
+ *
+ * We disable these timeout for the same reason as in the disable_client_timeout()
+ */
+extern void
+MtmDisableTimeouts(void)
+{
+	if (get_timeout_active(STATEMENT_TIMEOUT))
+	{
+		disable_timeout(STATEMENT_TIMEOUT, false);
+	}
+
+	if (get_timeout_active(LOCK_TIMEOUT))
+	{
+		disable_timeout(LOCK_TIMEOUT, false);
+	}
+
+	if (get_timeout_active(IDLE_IN_TRANSACTION_SESSION_TIMEOUT))
+	{
+		disable_timeout(IDLE_IN_TRANSACTION_SESSION_TIMEOUT, false);
+	}
+
+	if (get_timeout_active(IDLE_SESSION_TIMEOUT))
+	{
+		disable_timeout(IDLE_SESSION_TIMEOUT, false);
+	}
+}
+
+/*
+ * Wrapper on PQconnectPoll
+ *
+ * On connect disables timeouts on a client side
+ */
+PostgresPollingStatusType
+MtmPQconnectPoll(PGconn *conn)
+{
+	PostgresPollingStatusType status;
+
+	status = PQconnectPoll(conn);
+	if (status != PGRES_POLLING_OK)
+		return status;
+
+	disable_client_timeouts(conn);
+
+	return status;
+}
+
+/*
+ * Wrapper on PQconnectdb
+ *
+ * On connect disables timeouts on a client side
+ */
+PGconn *
+MtmPQconnectdb(const char *conninfo)
+{
+	PGconn *conn;
+
+	conn = PQconnectdb(conninfo);
+	if (PQstatus(conn) != CONNECTION_OK)
+		return conn;
+
+	disable_client_timeouts(conn);
+
+	return conn;
+}
+

src/multimaster.c

@@ -48,6 +48,7 @@
 #include "commit.h"
 #include "messaging.h"
 #include "syncpoint.h"
+#include "mtm_utils.h"
 
 #include "compat.h"
 
@@ -333,7 +334,6 @@ MtmSleep(int64 usec)
 	}
 }
 
-
 /*
  * These were once used to setup mtm state in parallel workers, but as long as
  * they are read-only we don't really need it (historically it imported csn
@@ -970,7 +970,7 @@ mtm_init_cluster(PG_FUNCTION_ARGS)
 		int			j;
 
 		/* connect */
-		peer_conns[i] = PQconnectdb(conninfo);
+		peer_conns[i] = MtmPQconnectdb(conninfo);
 		if (PQstatus(peer_conns[i]) != CONNECTION_OK)
 		{
 			char	   *msg = pchomp(PQerrorMessage(peer_conns[i]));
@@ -1300,7 +1300,7 @@ mtm_join_node(PG_FUNCTION_ARGS)
 	if (new_node == NULL)
 		mtm_log(ERROR, "new node %d not found", new_node_id);
 	conninfo = new_node->conninfo;
-	conn = PQconnectdb(conninfo);
+	conn = MtmPQconnectdb(conninfo);
 	if (PQstatus(conn) != CONNECTION_OK)
 	{
 		char	   *msg = pchomp(PQerrorMessage(conn));
@@ -1495,7 +1495,7 @@ mtm_ping(PG_FUNCTION_ARGS)
 		if (!BIT_CHECK(curr_gen.members, peer->node_id - 1))
 			continue;
 
-		conn = PQconnectdb(peer->conninfo);
+		conn = MtmPQconnectdb(peer->conninfo);
 		if (PQstatus(conn) != CONNECTION_OK)
 		{
 			char	   *msg = pchomp(PQerrorMessage(conn));
@@ -2554,7 +2554,7 @@ _mtm_get_snapshots(const MtmConfig *mcfg, PGconn **conns, char **snapnames,
 	for (i = 0; i < mcfg->n_nodes; i++)
 	{
 		/* Establish connection to each node */
-		conns[i] = PQconnectdb(mcfg->nodes[i].conninfo);
+		conns[i] = MtmPQconnectdb(mcfg->nodes[i].conninfo);
 
 		if (conns[i] == NULL || PQstatus(conns[i]) == CONNECTION_BAD)
 		{
@@ -2680,7 +2680,7 @@ mtm_check_query(PG_FUNCTION_ARGS)
 			int pos = index[i];
 
 			/* Establish connection to each online node */
-			conn = PQconnectdb(cfg->nodes[pos].conninfo);
+			conn = MtmPQconnectdb(cfg->nodes[pos].conninfo);
 
 			if (conn == NULL || PQstatus(conn) == CONNECTION_BAD)
 			{

src/pglogical_output.c

@@ -57,6 +57,7 @@
 #include "multimaster.h"
 #include "logger.h"
 #include "state.h"
+#include "mtm_utils.h"
 
 extern void _PG_output_plugin_init(OutputPluginCallbacks *cb);
 
@@ -143,6 +144,8 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 	cb->shutdown_cb = pg_decode_shutdown;
 	cb->message_cb = pg_decode_message;
 	cb->caughtup_cb = pg_decode_caughtup;
+
+	MtmDisableTimeouts();
 }
 
 #if 0

src/pglogical_receiver.c

@@ -56,6 +56,7 @@
 #include "compat.h"
 #include "syncpoint.h"
 #include "global_tx.h"
+#include "mtm_utils.h"
 
 #define ERRCODE_DUPLICATE_OBJECT_STR  "42710"
 
@@ -584,6 +585,8 @@ pglogical_receiver_main(Datum main_arg)
 	 */
 	on_shmem_exit(pglogical_receiver_at_exit, PointerGetDatum(rctx));
 
+	MtmDisableTimeouts();
+
 	MtmIsReceiver = true;
 	/* Run as replica session replication role. */
 	SetConfigOption("session_replication_role", "replica",

src/resolver.c

@@ -30,6 +30,7 @@
 #include "commit.h"
 #include "global_tx.h"
 #include "messaging.h"
+#include "mtm_utils.h"
 
 static MtmConfig *mtm_cfg = NULL;
 static bool		send_requests;
@@ -637,6 +638,8 @@ ResolverMain(Datum main_arg)
 	Oid			db_id,
 				user_id;
 
+	MtmDisableTimeouts();
+
 	/* init this worker */
 	pqsignal(SIGHUP, ResolverSigHupHandler);
 	pqsignal(SIGTERM, die);

src/state.c

@@ -45,6 +45,7 @@
 #include "syncpoint.h"
 #include "logger.h"
 #include "messaging.h"
+#include "mtm_utils.h"
 
 char const *const MtmNeighborEventMnem[] =
 {
@@ -1672,6 +1673,8 @@ CampaignerMain(Datum main_arg)
 	TimestampTz last_campaign_at = 0;
 	int			rc = WL_TIMEOUT;
 
+	MtmDisableTimeouts();
+
 	MtmBackgroundWorker = true;
 	mtm_log(MtmStateMessage, "campaigner started");
 	before_shmem_exit(CampaignerOnExit, (Datum) 0);
@@ -3417,6 +3420,8 @@ ReplierMain(Datum main_arg)
 													  ALLOCSET_DEFAULT_SIZES);
 	bool	job_pending;
 
+	MtmDisableTimeouts();
+
 	MtmBackgroundWorker = true;
 	before_shmem_exit(ReplierOnExit, (Datum) 0);
 	mtm_log(MtmStateMessage, "replier started");