Clean referee decision when all nodes are online; do not send requests to referee continously from disabled node

Author: kelvich

multimaster--1.0.sql

@@ -170,8 +170,11 @@ END
 $$
 LANGUAGE plpgsql;
 
-CREATE OR REPLACE FUNCTION mtm.referee_clean() RETURNS void AS
+CREATE OR REPLACE FUNCTION mtm.referee_clean() RETURNS bool AS
 $$
+BEGIN
     delete from mtm.referee_decision where key = 'winner';
+    return 'true';
+END
 $$
-LANGUAGE sql;
+LANGUAGE plpgsql;

multimaster.c

@@ -2376,6 +2376,7 @@ static void MtmInitialize()
 		Mtm->disabledNodeMask =  (((nodemask_t)1 << MtmNodes) - 1);
 		Mtm->clique = 0;
 		Mtm->refereeGrant = false;
+		Mtm->refereeWinnerId = 0;
 		Mtm->stalledNodeMask = 0;
 		Mtm->stoppedNodeMask = 0;
 		Mtm->deadNodeMask = 0;
@@ -4158,15 +4159,42 @@ static void erase_option_from_connstr(const char *option, char *connstr)
 	pfree(needle);
 }
 
-PGconn *PQconnectdb_safe(const char *conninfo)
+PGconn *
+PQconnectdb_safe(const char *conninfo, int timeout)
 {
 	PGconn *conn;
+	struct timeval tv = { timeout, 0 };
 	char *safe_connstr = pstrdup(conninfo);
-	erase_option_from_connstr("arbiter_port", safe_connstr);
 
-	conn = PQconnectdb(safe_connstr);
+	/* XXXX add timeout to connstring if set */
 
+	erase_option_from_connstr("arbiter_port", safe_connstr);
+	conn = PQconnectdb(safe_connstr);
 	pfree(safe_connstr);
+
+	if (PQstatus(conn) != CONNECTION_OK)
+	{
+		MTM_ELOG(WARNING, "Could not connect to '%s': %s",
+			safe_connstr, PQerrorMessage(conn));
+	}
+
+	if (timeout != 0)
+	{
+		int socket_fd = PQsocket(conn);
+
+		if (socket_fd < 0)
+		{
+			MTM_ELOG(WARNING, "Referee socket is invalid");
+		}
+
+		if (setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO,
+									(char *)&tv, sizeof(tv)) < 0)
+		{
+			MTM_ELOG(WARNING, "Could not set referee socket timeout: %s",
+						strerror(errno));
+		}
+	}
+
 	return conn;
 }
 
@@ -4202,7 +4230,7 @@ mtm_collect_cluster_info(PG_FUNCTION_ARGS)
 		SRF_RETURN_DONE(funcctx);
 	}
 
-	conn = PQconnectdb_safe(Mtm->nodes[usrfctx->nodeId-1].con.connStr);
+	conn = PQconnectdb_safe(Mtm->nodes[usrfctx->nodeId-1].con.connStr, 0);
 	if (PQstatus(conn) != CONNECTION_OK)
 	{
 		MTM_ELOG(WARNING, "Failed to establish connection '%s' to node %d: error = %s", Mtm->nodes[usrfctx->nodeId-1].con.connStr, usrfctx->nodeId, PQerrorMessage(conn));
@@ -4411,7 +4439,7 @@ static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError, int force
 	{
 		if (!BIT_CHECK(disabledNodeMask, i) || (i + 1 == forceOnNode))
 		{
-			conns[i] = PQconnectdb_safe(psprintf("%s application_name=%s", Mtm->nodes[i].con.connStr, MULTIMASTER_BROADCAST_SERVICE));
+			conns[i] = PQconnectdb_safe(psprintf("%s application_name=%s", Mtm->nodes[i].con.connStr, MULTIMASTER_BROADCAST_SERVICE), 0);
 			if (PQstatus(conns[i]) != CONNECTION_OK)
 			{
 				if (ignoreError)

multimaster.h

@@ -289,7 +289,8 @@ typedef struct
 	TransactionId oldestXid;           /* XID of oldest transaction visible by any active transaction (local or global) */
 	nodemask_t disabledNodeMask;       /* Bitmask of disabled nodes */
 	nodemask_t clique;                 /* Bitmask of nodes that are connected and we allowed to connect/send wal/receive wal with them */
-	bool refereeGrant;                 /* Referee allowed us to work with half of the nodes */
+	bool       refereeGrant;           /* Referee allowed us to work with half of the nodes */
+	int        refereeWinnerId;        /* Node that won referee contest */
 	nodemask_t deadNodeMask;           /* Bitmask of nodes considered as dead by referee */
 	nodemask_t recoveredNodeMask;      /* Bitmask of nodes recoverd after been reported as dead by referee */
 	nodemask_t stalledNodeMask;        /* Bitmask of stalled nodes (node with dropped replication slot which makes it not possible automatic recovery of such node) */
@@ -430,7 +431,7 @@ extern void MtmCheckHeartbeat(void);
 extern void MtmResetTransaction(void);
 extern void MtmUpdateLockGraph(int nodeId, void const* messageBody, int messageSize);
 extern void MtmReleaseRecoverySlot(int nodeId);
-extern PGconn *PQconnectdb_safe(const char *conninfo);
+extern PGconn *PQconnectdb_safe(const char *conninfo, int timeout);
 extern void MtmBeginSession(int nodeId);
 extern void MtmEndSession(int nodeId, bool unlock);
 extern void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit);

pglogical_receiver.c

@@ -296,7 +296,7 @@ pglogical_receiver_main(Datum main_arg)
 		count = Mtm->recoveryCount;
 
 		/* Establish connection to remote server */
-		conn = PQconnectdb_safe(connString);
+		conn = PQconnectdb_safe(connString, 0);
 		status = PQstatus(conn);
 		if (status != CONNECTION_OK)
 		{

referee.c

@@ -50,7 +50,7 @@ static void MtmRefereeLoop(char const** connections, int nConns)
 	int result;
 
 	for (i = 0; i < nConns; i++) {
-		conns[i] = PQconnectdb_safe(connections[i]);
+		conns[i] = PQconnectdb_safe(connections[i], 0);
 		status = PQstatus(conns[i]);
 		if (status != CONNECTION_OK)
 		{
@@ -74,7 +74,7 @@ static void MtmRefereeLoop(char const** connections, int nConns)
 			/* Some of live node reestablished connection with dead node, so referee should also try to connect to this node */
 			if (conns[i] == NULL) { 
 				if (BIT_CHECK(newEnabledMask, i)) { 
-					conns[i] = PQconnectdb_safe(connections[i]);
+					conns[i] = PQconnectdb_safe(connections[i], 0);
 					status = PQstatus(conns[i]);
 					if (status == CONNECTION_OK) {
 						BIT_CLEAR(disabledMask, i);

state.c

@@ -25,7 +25,8 @@ char const* const MtmEventMnem[] =
 	"MTM_NONRECOVERABLE_ERROR"
 };
 
-static int MtmGetRefereeWinner(void);
+static int  MtmRefereeGetWinner(void);
+static bool MtmRefereeClearWinner(void);
 
 // XXXX: allocate in context and clean it
 static char *
@@ -81,7 +82,7 @@ MtmCheckState(void)
 	// int nVotingNodes = MtmGetNumberOfVotingNodes();
 	bool isEnabledState;
 	int nEnabled   = countZeroBits(Mtm->disabledNodeMask, Mtm->nAllNodes);
-	int nConnected = countZeroBits(SELF_CONNECTIVITY_MASK, Mtm->nAllNodes);
+	int nConnected = countZeroBits(EFFECTIVE_CONNECTIVITY_MASK, Mtm->nAllNodes);
 	int nReceivers = Mtm->nAllNodes - countZeroBits(Mtm->pglogicalReceiverMask, Mtm->nAllNodes);
 	int nSenders   = Mtm->nAllNodes - countZeroBits(Mtm->pglogicalSenderMask, Mtm->nAllNodes);
 
@@ -390,26 +391,46 @@ MtmRefreshClusterStatus()
 	MtmCheckState();
 
 	/*
-	 * Check for referee decidion when pnly half of nodes are visible.
+	 * Check for referee decision when only half of nodes are visible.
 	 */
-	if (MtmRefereeConnStr && *MtmRefereeConnStr && !Mtm->refereeGrant &&
+	if (MtmRefereeConnStr && *MtmRefereeConnStr && !Mtm->refereeWinnerId &&
 		countZeroBits(EFFECTIVE_CONNECTIVITY_MASK, Mtm->nAllNodes) == Mtm->nAllNodes/2)
 	{
-		int winner_node_id = MtmGetRefereeWinner();
-		if (winner_node_id != -1 &&
-			!BIT_CHECK(EFFECTIVE_CONNECTIVITY_MASK, winner_node_id - 1))
+		int winner_node_id = MtmRefereeGetWinner();
+
+		if (winner_node_id > 0)
 		{
-			MTM_LOG1("[STATE] Referee allowed to proceed with half of the nodes (winner_id = %d)",
-						winner_node_id);
-			Mtm->refereeGrant = true;
-
-			MtmLock(LW_EXCLUSIVE);
-			MtmEnableNode(MtmNodeId);
-			MtmCheckState();
-			MtmUnlock();
+			Mtm->refereeWinnerId = winner_node_id;
+			if (!BIT_CHECK(EFFECTIVE_CONNECTIVITY_MASK, winner_node_id - 1))
+			{
+				MTM_LOG1("[STATE] Referee allowed to proceed with half of the nodes (winner_id = %d)",
+							winner_node_id);
+				Mtm->refereeGrant = true;
+				MtmLock(LW_EXCLUSIVE);
+				MtmEnableNode(MtmNodeId);
+				MtmCheckState();
+				MtmUnlock();
+			}
 		}
 	}
 
+	/*
+	 * Clear winner if we again have all nodes are online.
+	 */
+	if (MtmRefereeConnStr && *MtmRefereeConnStr && Mtm->refereeWinnerId &&
+		countZeroBits(EFFECTIVE_CONNECTIVITY_MASK, Mtm->nAllNodes) == Mtm->nAllNodes)
+	{
+		if (MtmRefereeClearWinner())
+		{
+			Mtm->refereeWinnerId = 0;
+			Mtm->refereeGrant = false;
+			MTM_LOG1("[STATE] Cleaning old referee decision");
+		}
+	}
+
+	/* Do not check clique with referee grant */
+	if (Mtm->refereeGrant)
+		return;
 
 	/*
 	 * Check for clique.
@@ -473,37 +494,17 @@ MtmRefreshClusterStatus()
 }
 
 static int
-MtmGetRefereeWinner(void)
+MtmRefereeGetWinner(void)
 {
-	int socket_fd;
 	PGconn* conn;
 	PGresult *res;
-	struct timeval timeout = { 5, 0 };
 	char sql[128];
 	int  winner_node_id;
 
-	conn = PQconnectdb_safe(MtmRefereeConnStr);
+	conn = PQconnectdb_safe(MtmRefereeConnStr, 5);
 	if (PQstatus(conn) != CONNECTION_OK)
 	{
-		MTM_ELOG(WARNING, "Could not connect to referee (%s): %s",
-					MtmRefereeConnStr, PQerrorMessage(conn));
-		PQfinish(conn);
-		return -1;
-	}
-
-	socket_fd = PQsocket(conn);
-	if (socket_fd < 0)
-	{
-		MTM_ELOG(WARNING, "Referee socket is invalid");
-		PQfinish(conn);
-		return -1;
-	}
-
-	if (setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO,
-								(char *)&timeout, sizeof(timeout)) < 0)
-	{
-		MTM_ELOG(WARNING, "Could not set referee socket timeout: %s",
-					strerror(errno));
+		MTM_ELOG(WARNING, "Could not connect to referee");
 		PQfinish(conn);
 		return -1;
 	}
@@ -540,3 +541,46 @@ MtmGetRefereeWinner(void)
 	return winner_node_id;
 }
 
+static bool
+MtmRefereeClearWinner(void)
+{
+	PGconn* conn;
+	PGresult *res;
+	char *response;
+
+	conn = PQconnectdb_safe(MtmRefereeConnStr, 5);
+	if (PQstatus(conn) != CONNECTION_OK)
+	{
+		MTM_ELOG(WARNING, "Could not connect to referee");
+		PQfinish(conn);
+		return false;
+	}
+
+	res = PQexec(conn, "select mtm.referee_clean()");
+	if (PQresultStatus(res) != PGRES_TUPLES_OK ||
+		PQntuples(res) != 1 ||
+		PQnfields(res) != 1)
+	{
+		MTM_ELOG(WARNING, "Refusing unexpected result (r=%d, n=%d, w=%d, k=%s) from referee_clean().",
+			PQresultStatus(res), PQntuples(res), PQnfields(res), PQgetvalue(res, 0, 0));
+		PQclear(res);
+		PQfinish(conn);
+		return false;
+	}
+
+	response = PQgetvalue(res, 0, 0);
+
+	if (false)
+	{
+		MTM_ELOG(WARNING, "Wrong response from referee");
+		PQclear(res);
+		PQfinish(conn);
+		return false;
+	}
+
+	/* Ok, we finally got it! */
+	MTM_LOG1("Got referee clear response %s", response);
+	PQclear(res);
+	PQfinish(conn);
+	return true;
+}

tests2/lib/test_helper.py

@@ -34,7 +34,7 @@ def assertNoCommits(self, aggs):
         if commits:
             raise AssertionError('There are commits during aggregation interval')
 
-    def performFailure(self, failure):
+    def performFailure(self, failure, wait=0):
 
         time.sleep(TEST_WARMING_TIME)
 
@@ -51,7 +51,7 @@ def performFailure(self, failure):
         aggs_failure = self.client.get_aggregates()
 
 
-        # time.sleep(10000)
+        time.sleep(wait)
         failure.stop()
 
         print('Eliminate failure at ',datetime.datetime.utcnow())

tests2/test_major.py

@@ -95,6 +95,28 @@ def test_partition_referee(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
+    def test_double_failure_referee(self):
+        print('### test_double_failure_referee ###')
+
+        aggs_failure, aggs = self.performFailure(SingleNodePartition('node2'))
+
+        self.assertCommits(aggs_failure[:1])
+        self.assertNoCommits(aggs_failure[1:])
+        self.assertIsolation(aggs_failure)
+
+        self.assertCommits(aggs)
+        self.assertIsolation(aggs)
+
+        aggs_failure, aggs = self.performFailure(SingleNodePartition('node1'))
+
+        self.assertNoCommits(aggs_failure[:1])
+        self.assertCommits(aggs_failure[1:])
+        self.assertIsolation(aggs_failure)
+
+        self.assertCommits(aggs)
+        self.assertIsolation(aggs)
+
+
 if __name__ == '__main__':
     unittest.main()