referee version 2

Author: kelvich

multimaster--1.0.sql

@@ -151,3 +151,27 @@ $$
 LANGUAGE plpgsql;
 
 -- select mtm.alter_sequences();
+
+-- referee stuff
+CREATE TABLE IF NOT EXISTS mtm.referee_decision(key text primary key not null, node_id int);
+
+CREATE OR REPLACE FUNCTION mtm.referee_get_winner(applicant_id int) RETURNS int AS
+$$
+DECLARE
+    winner_id int;
+BEGIN
+    insert into mtm.referee_decision values ('winner', applicant_id);
+    select node_id into winner_id from mtm.referee_decision where key = 'winner';
+    return winner_id;
+EXCEPTION WHEN others THEN
+    select node_id into winner_id from mtm.referee_decision where key = 'winner';
+    return winner_id;
+END
+$$
+LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION mtm.referee_clean() RETURNS void AS
+$$
+    delete from mtm.referee_decision where key = 'winner';
+$$
+LANGUAGE sql;

multimaster.c

@@ -254,6 +254,7 @@ bool  MtmUseRDMA;
 bool  MtmPreserveCommitOrder;
 bool  MtmVolksWagenMode; /* Pretend to be normal postgres. This means skip some NOTICE's and use local sequences */
 bool  MtmMajorNode;
+char* MtmRefereeConnStr;
 
 static char* MtmConnStrs;
 static char* MtmRemoteFunctionsList;
@@ -2372,6 +2373,7 @@ static void MtmInitialize()
 		Mtm->nAllNodes = MtmNodes;
 		Mtm->disabledNodeMask =  (((nodemask_t)1 << MtmNodes) - 1);
 		Mtm->clique = 0;
+		Mtm->refereeGrant = false;
 		Mtm->stalledNodeMask = 0;
 		Mtm->stoppedNodeMask = 0;
 		Mtm->deadNodeMask = 0;
@@ -2618,8 +2620,7 @@ static void MtmSplitConnStrs(void)
 		}
 		pfree(copy);
 	}
-	if (!MtmReferee)
-	{
+
 		if (MtmNodeId == INT_MAX) {
 			if (gethostname(buf, sizeof buf) != 0) {
 				MTM_ELOG(ERROR, "Failed to get host name: %m");
@@ -2679,7 +2680,6 @@ static void MtmSplitConnStrs(void)
 			len = end - dbName;
 			MtmDatabaseName = pnstrdup(dbName, len);
 		}
-	}
 	MemoryContextSwitchTo(old_context);
 }
 
@@ -2976,6 +2976,19 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomStringVariable(
+		"multimaster.referee_connstring",
+		"Referee connection string",
+		NULL,
+		&MtmRefereeConnStr,
+		"",
+		PGC_POSTMASTER,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomBoolVariable(
 		"multimaster.use_rdma",
 		"Use RDMA sockets",
@@ -3177,8 +3190,6 @@ _PG_init(void)
 
 	if (MtmReferee)
 	{
-		MtmSplitConnStrs();
-		MtmRefereeInitialize();
 		return;
 	}
 

multimaster.h

@@ -287,6 +287,7 @@ 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 */
 	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) */
@@ -379,6 +380,7 @@ extern timestamp_t MtmRefreshClusterStatusSchedule;
 extern MtmConnectionInfo* MtmConnections;
 extern bool MtmMajorNode;
 extern bool MtmBackgroundWorker;
+extern char* MtmRefereeConnStr;
 
 
 extern void  MtmArbiterInitialize(void);

state.c

@@ -25,6 +25,7 @@ char const* const MtmEventMnem[] =
 	"MTM_NONRECOVERABLE_ERROR"
 };
 
+static int MtmGetRefereeWinner(void);
 
 // XXXX: allocate in context and clean it
 static char *
@@ -91,12 +92,14 @@ MtmCheckState(void)
 		maskToString(Mtm->pglogicalReceiverMask, Mtm->nAllNodes),
 		maskToString(Mtm->pglogicalSenderMask, Mtm->nAllNodes),
 		Mtm->nAllNodes,
-		MtmMajorNode);
+		(MtmMajorNode || Mtm->refereeGrant) );
 
 	isEnabledState =
-		( (nConnected >= Mtm->nAllNodes/2+1)						/* majority */
-			|| (nConnected == Mtm->nAllNodes/2 && MtmMajorNode) )	/* or half + major node */
-		&& BIT_CHECK(Mtm->clique, MtmNodeId-1);						/* in clique */
+		( (nConnected >= Mtm->nAllNodes/2+1)							/* majority */
+			// XXXX: should we restrict major with two nodes setup?
+			|| (nConnected == Mtm->nAllNodes/2 && MtmMajorNode)			/* or half + major node */
+			|| (nConnected == Mtm->nAllNodes/2 && Mtm->refereeGrant) )  /* or half + referee */
+		&& BIT_CHECK(Mtm->clique, MtmNodeId-1);							/* in clique */
 
 	/* ANY -> MTM_DISABLED */
 	if (!isEnabledState)
@@ -135,6 +138,12 @@ MtmCheckState(void)
 		case MTM_RECOVERED:
 			if (nReceivers == nEnabled-1 && nSenders == nEnabled-1 && nEnabled == nConnected)
 			{
+				/*
+				 * It should be already cleaned by RECOVERY_CAUGHTUP, but
+				 * in major mode or with referee we can be working alone
+				 * so nobody will clean it.
+				 */
+				BIT_CLEAR(Mtm->originLockNodeMask, MtmNodeId-1);
 				MtmSetClusterStatus(MTM_ONLINE);
 				return;
 			}
@@ -376,9 +385,35 @@ MtmRefreshClusterStatus()
 	 * Periodical check that we are still in RECOVERED state.
 	 * See comment to MTM_RECOVERED -> MTM_ONLINE transition in MtmCheckState()
 	 */
-	if (Mtm->status == MTM_RECOVERED)
-		MtmCheckState();
+	MtmCheckState();
+
+	/*
+	 * Check for referee decidion when pnly half of nodes are visible.
+	 */
+	if (MtmRefereeConnStr && *MtmRefereeConnStr && !Mtm->refereeGrant &&
+		// XXXX visibility & ~clique?
+		countZeroBits(SELF_CONNECTIVITY_MASK, Mtm->nAllNodes) == Mtm->nAllNodes/2)
+	{
+		int winner_node_id = MtmGetRefereeWinner();
+		if (winner_node_id != -1 &&
+			// XXXX visibility & ~clique?
+			!BIT_CHECK(SELF_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();
+		}
+	}
 
+
+	/*
+	 * Check for clique.
+	 */
 	MtmBuildConnectivityMatrix(matrix);
 	newClique = MtmFindMaxClique(matrix, Mtm->nAllNodes, &cliqueSize);
 
@@ -436,3 +471,70 @@ MtmRefreshClusterStatus()
 	MtmCheckState();
 	MtmUnlock();
 }
+
+static int
+MtmGetRefereeWinner(void)
+{
+	int socket_fd;
+	PGconn* conn;
+	PGresult *res;
+	struct timeval timeout = { 5, 0 };
+	char sql[128];
+	int  winner_node_id;
+
+	conn = PQconnectdb_safe(MtmRefereeConnStr);
+	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));
+		PQfinish(conn);
+		return -1;
+	}
+
+	sprintf(sql, "select mtm.referee_get_winner(%d)", MtmNodeId);
+	res = PQexec(conn, sql);
+	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.",
+			PQresultStatus(res), PQntuples(res), PQnfields(res), PQgetvalue(res, 0, 0));
+		PQclear(res);
+		PQfinish(conn);
+		return -1;
+	}
+
+	winner_node_id = atoi(PQgetvalue(res, 0, 0));
+
+	if (winner_node_id < 1 || winner_node_id > Mtm->nAllNodes)
+	{
+		MTM_ELOG(WARNING,
+			"Referee responded with node_id=%d, it's out of our node range",
+			winner_node_id);
+		PQclear(res);
+		PQfinish(conn);
+		return -1;
+	}
+
+	MTM_LOG1("Got referee response, winner node_id=%d.", winner_node_id);
+	/* Ok, we finally got it! */
+	return winner_node_id;
+}
+

tests2/docker-entrypoint.sh

@@ -58,21 +58,39 @@ if [ "$1" = 'postgres' ]; then
 			max_wal_senders = 10
 			shared_preload_libraries = 'multimaster'
 			default_transaction_isolation = 'repeatable read'
-            log_line_prefix = '%m: '
+			log_line_prefix = '%m: '
+			# log_statement = all
 
 			multimaster.workers = 4
 			multimaster.max_workers = 16
 			multimaster.max_nodes = 3
 			multimaster.volkswagen_mode = 1
 			multimaster.queue_size=52857600
 			multimaster.ignore_tables_without_pk = 1
-			multimaster.node_id = $NODE_ID
-			multimaster.conn_strings = '$CONNSTRS'
-			multimaster.major_node = $MAJOR
 			multimaster.heartbeat_recv_timeout = 1100
 			multimaster.heartbeat_send_timeout = 250
 		EOF
 
+		if [ -n "$NODE_ID" ]; then
+			echo "multimaster.node_id = $NODE_ID" >> $PGDATA/postgresql.conf
+		fi
+
+		if [ -n "$CONNSTRS" ]; then
+			echo "multimaster.conn_strings = '$CONNSTRS'" >> $PGDATA/postgresql.conf
+		fi
+
+		if [ -n "$MAJOR" ]; then
+			echo 'multimaster.major_node = on' >> $PGDATA/postgresql.conf
+		fi
+
+		if [ -n "$REFEREE" ]; then
+			echo 'multimaster.referee = on' >> $PGDATA/postgresql.conf
+		fi
+
+		if [ -n "$REFEREE_CONNSTR" ]; then
+			echo "multimaster.referee_connstring = '$REFEREE_CONNSTR'" >> $PGDATA/postgresql.conf
+		fi
+
 		cat $PGDATA/postgresql.conf
 
 		pg_ctl -D "$PGDATA" -m fast -w stop

tests2/lib/test_helper.py

@@ -1,6 +1,7 @@
 import unittest
 import time
 import datetime
+import psycopg2
 
 TEST_WARMING_TIME = 5
 TEST_DURATION = 10
@@ -50,6 +51,7 @@ def performFailure(self, failure):
         aggs_failure = self.client.get_aggregates()
 
 
+        # time.sleep(10000)
         failure.stop()
 
         print('Eliminate failure at ',datetime.datetime.utcnow())
@@ -59,3 +61,12 @@ def performFailure(self, failure):
         aggs = self.client.get_aggregates()
 
         return (aggs_failure, aggs)
+
+    def nodeExecute(dsn, statements):
+        con = psycopg2.connect(dsn)
+        con.autocommit = True
+        cur = con.cursor()
+        for statement in statements:
+            cur.execute(statement)
+        cur.close()
+        con.close()

tests2/support/two_nodes.yml

@@ -12,10 +12,10 @@ services:
       POSTGRES_USER: 'pg'
       POSTGRES_DB: 'regression'
       NODE_ID: 1
-      MAJOR: 'off'
       CONNSTRS: >-
         dbname=regression user=pg host=node1,
         dbname=regression user=pg host=node2
+      REFEREE_CONNSTR: 'dbname=regression user=pg host=referee'
     ports:
       - "15432:5432"
 
@@ -29,9 +29,23 @@ services:
       POSTGRES_USER: 'pg'
       POSTGRES_DB: 'regression'
       NODE_ID: 2
-      MAJOR: 'off'
       CONNSTRS: >-
         dbname=regression user=pg host=node1,
         dbname=regression user=pg host=node2
+      REFEREE_CONNSTR: 'dbname=regression user=pg host=referee'
     ports:
       - "15433:5432"
+
+  referee:
+    container_name: referee
+    build: ../..
+    privileged: true
+    ulimits:
+      core: 14294967296
+    environment:
+      POSTGRES_USER: 'pg'
+      POSTGRES_DB: 'regression'
+      NODE_ID: 1
+      REFEREE: 'on'
+    ports:
+      - "15435:5432"