Fix fragile issue of DMQ sender service. Restart Campaigner at the DMQ sender service exit. Do it to ensure that critical data are cleaned before next voting.

Tags: Multimaster

Author: danolivo

src/dmq.c

@@ -32,6 +32,8 @@
 #include "logger.h"
 #include "compat.h"
 #include "mtm_utils.h"
+#include "multimaster.h"
+#include "state.h"
 
 #include "access/transam.h"
 #include "libpq/libpq.h"
@@ -52,23 +54,6 @@
 #define DMQ_MQ_SIZE  ((Size) 65536)
 #define DMQ_MQ_MAGIC 0x646d71
 
-/*  XXX: move to common */
-#define BIT_CLEAR(mask, bit) ((mask) &= ~((uint64)1 << (bit)))
-#define BIT_CHECK(mask, bit) (((mask) & ((uint64)1 << (bit))) != 0)
-static int
-first_set_bit(uint64 mask)
-{
-	int i;
-
-	for (i = 0; i < DMQ_N_MASK_POS; i++)
-	{
-		if (BIT_CHECK(mask, i))
-			return i;
-	}
-	return -1;
-}
-
-
 /*
  * Shared data structures to hold current connections topology.
  * All that stuff can be moved to persistent tables to avoid hardcoded
@@ -497,6 +482,12 @@ dmq_sender_at_exit(int status, Datum arg)
 		}
 	}
 	LWLockRelease(dmq_state->lock);
+
+	/*
+	 * Restart the Campaigner to be sure that all critical data reset before the
+	 * next voting.
+	 */
+	CampaignerStop();
 }
 
 void
@@ -1390,7 +1381,7 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 	void		*extra = NULL;
 
 	/*
-	 * We do not call MtmDisbaleTimeouts() here because of connection to this
+	 * We do not call MtmDisableTimeouts() here because of connection to this
 	 * client is made by MtmPQconnectPoll() that sets all needed timeouts.
 	 */
 
@@ -1660,7 +1651,7 @@ dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *payload
 	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, false);
 	pfree(buf.data);
 	if (res != SHM_MQ_SUCCESS)
-		mtm_log(WARNING, "[DMQ] dmq_push: can't send to queue");
+		mtm_log(ERROR, "[DMQ] dmq_push: can't send to queue");
 }
 
 /*
@@ -1776,7 +1767,8 @@ dmq_reattach_shm_mq(int handle_id)
  * from which receivers caller wants to get message and filters inhandles
  * through it.
  */
-void dmq_attach_receiver(char *sender_name, int8 mask_pos)
+void
+dmq_attach_receiver(char *sender_name, int8 mask_pos)
 {
 	int			i;
 	int			handle_id = -1;

src/include/state.h

@@ -149,5 +149,6 @@ extern void MtmMonitorStart(Oid db_id, Oid user_id);
 extern void MtmRefreshClusterStatus(void);
 extern nodemask_t MtmGetDisabledNodeMask(void);
 extern nodemask_t MtmGetEnabledNodeMask(bool ignore_disabled);
+extern void CampaignerStop(void);
 
 #endif

src/mtm_utils.c

@@ -22,42 +22,36 @@
  *
  * 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. 
+ * these timeouts.
  *
  * This function raises an error on PQExec failed.
  */
-static void
+static bool
 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);
-	}
+		mtm_log(ERROR, "failed to set statement_timeout: %s",
+				pchomp(PQerrorMessage(conn)));
 
 	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);
-	}
+		mtm_log(ERROR, "failed to set lock_timeout: %s",
+				pchomp(PQerrorMessage(conn)));
 
 	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);
-	}
+		mtm_log(ERROR, "failed to set idle_in_transaction_session_timeout: %s",
+				pchomp(PQerrorMessage(conn)));
 
 	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);
-	}
+		mtm_log(ERROR, "failed to set idle_session_timeout: %s",
+				pchomp(PQerrorMessage(conn)));
+
+	return true;
 }
 
 /*
@@ -73,24 +67,13 @@ 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);
-	}
 }
 
 /*

src/multimaster.c

@@ -99,7 +99,6 @@ static void MtmShmemStartup(void);
 
 static void launcher_init(void);
 void		launcher_main(Datum main_arg);
-void		drop_node_entry(int node_id);
 
 MtmShared  *Mtm;
 
@@ -1863,6 +1862,7 @@ launcher_main(Datum main_arg)
 	/* init this worker */
 	pqsignal(SIGTERM, die);
 	BackgroundWorkerUnblockSignals();
+	MtmBackgroundWorker = true;
 
 	memset(&hash_info, 0, sizeof(hash_info));
 	hash_info.entrysize = hash_info.keysize = sizeof(Oid);

src/state.c

@@ -882,6 +882,14 @@ CampaignerWake(void)
 		kill(mtm_state->campaigner_pid, SIGHUP);
 }
 
+/* Service to restart a campaigner process. */
+void
+CampaignerStop(void)
+{
+	if (mtm_state->campaigner_pid != 0)
+		kill(mtm_state->campaigner_pid, SIGTERM);
+}
+
 /* campaigner never rereads PG config, but it currently it hardly needs to */
 static void
 CampaignerSigHupHandler(SIGNAL_ARGS)
@@ -3856,6 +3864,12 @@ stop_node_workers(int node_id, MtmConfig *new_cfg, Datum arg)
 	pfree(bgws[node_id - 1].handle);
 	bgws[node_id - 1].handle = NULL;
 
+	/*
+	 * Only cleaning a name field guarantees that monitor wouldn't restart this
+	 * receiver.
+	*/
+	bgws[node_id - 1].name[0] = '\0';
+
 	/* delete recovery slot, was acquired by receiver */
 	ReplicationSlotDrop(filter_slot_name, true);
 
@@ -4102,6 +4116,8 @@ MtmMonitor(Datum arg)
 			 * Note that if for some reason monitor wasn't running
 			 * (e.g. process killed) during node drop, cleanup in
 			 * stop_node_workers will be skipped. Very unlikely, but not nice.
+			 * XXX: Should rethink this code, because problem, described above,
+			 * has cought in PGPRO-6146.
 			 */
 			mtm_cfg = MtmReloadConfig(mtm_cfg, start_node_workers,
 									  stop_node_workers, PointerGetDatum(bgws),