Revert "Merge branch 'PGPROEE9_6' into PGPROEE9_6_MULTIMASTER"

This reverts commit 8581033, reversing
changes made to 573beb1.

Revert multimaster to state of release 9.6.5.1

Author: vbwagner

contrib/Makefile

@@ -47,7 +47,6 @@ SUBDIRS = \
 		pg_visibility	\
 		pg_wait_sampling \
 		postgres_fdw	\
-		referee	\
 		rum		\
 		seg		\
 		spi		\

contrib/mmts/Cluster.pm

@@ -66,7 +66,6 @@ sub new
 		my $node = new PostgresNode("node$i", $host, $pgport);
 		$node->{id} = $i;
 		$node->{arbiter_port} = $arbiter_port;
-		$node->{mmconnstr} = "${ \$node->connstr('postgres') } arbiter_port=${ \$node->{arbiter_port} }";
 		push(@$nodes, $node);
 	}
 
@@ -90,54 +89,47 @@ sub init
 	}
 }
 
-sub all_connstrs
-{
-	my ($self) = @_;
-	my $nodes = $self->{nodes};
-	return join(', ', map { "${ \$_->connstr('postgres') } arbiter_port=${ \$_->{arbiter_port} }" } @$nodes);
-}
-
-
 sub configure
 {
 	my ($self) = @_;
 	my $nodes = $self->{nodes};
+	my $nnodes = scalar @{ $nodes };
 
-	my $connstr = $self->all_connstrs();
+	my $connstr = join(', ', map { "${ \$_->connstr('postgres') } arbiter_port=${ \$_->{arbiter_port} }" } @$nodes);
 
 	foreach my $node (@$nodes)
 	{
 		my $id = $node->{id};
 		my $host = $node->host;
 		my $pgport = $node->port;
 		my $arbiter_port = $node->{arbiter_port};
-		my $unix_sock_dir = $ENV{PGHOST};
 
 		$node->append_conf("postgresql.conf", qq(
 			log_statement = none
 			listen_addresses = '$host'
-			unix_socket_directories = '$unix_sock_dir'
+			unix_socket_directories = ''
 			port = $pgport
 			max_prepared_transactions = 10
 			max_connections = 10
 			max_worker_processes = 100
 			wal_level = logical
-			max_wal_senders = 6
+			max_wal_senders = 5
 			wal_sender_timeout = 0
 			default_transaction_isolation = 'repeatable read'
-			max_replication_slots = 6
+			max_replication_slots = 5
 			shared_preload_libraries = 'multimaster'
 			shared_buffers = 16MB
 
 			multimaster.arbiter_port = $arbiter_port
 			multimaster.workers = 1
 			multimaster.node_id = $id
 			multimaster.conn_strings = '$connstr'
-			multimaster.heartbeat_recv_timeout = 1050
+			multimaster.heartbeat_recv_timeout = 2050
 			multimaster.heartbeat_send_timeout = 250
-			multimaster.max_nodes = 6
-			multimaster.ignore_tables_without_pk = false
+			multimaster.max_nodes = $nnodes
+			multimaster.ignore_tables_without_pk = true
 			multimaster.queue_size = 4194304
+			multimaster.min_2pc_timeout = 150000
 			log_line_prefix = '%t: '
 		));
 

contrib/mmts/Dockerfile

@@ -6,9 +6,6 @@ COPY ./ /pg/mmts/
 RUN export USE_PGXS=1 && \
     cd /pg/mmts && make clean && make install
 
-RUN export USE_PGXS=1 && \
-    cd /pg/src/contrib/referee && make clean && make install
-
 # pg_regress client assumes such dir exists on server
 RUN cp /pg/src/src/test/regress/*.so /pg/install/lib/postgresql/
 USER postgres

contrib/mmts/Makefile

@@ -1,7 +1,7 @@
 
 EXTENSION = multimaster
 DATA = multimaster--1.0.sql
-OBJS = multimaster.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o referee.o state.o
+OBJS = multimaster.o arbiter.o bytebuf.o bgwpool.o pglogical_output.o pglogical_proto.o pglogical_receiver.o pglogical_apply.o pglogical_hooks.o pglogical_config.o pglogical_relid_map.o ddd.o bkb.o spill.o referee.o
 MODULE_big = multimaster
 
 PG_CPPFLAGS = -I$(libpq_srcdir)

contrib/mmts/arbiter.c

@@ -76,7 +76,6 @@
 
 
 #include "multimaster.h"
-#include "state.h"
 
 #define MAX_ROUTES       16
 #define INIT_BUFFER_SIZE 1024
@@ -190,6 +189,7 @@ static void MtmDisconnect(int node)
 	MtmUnregisterSocket(sockets[node]);
 	pg_closesocket(sockets[node], MtmUseRDMA);
 	sockets[node] = -1;
+	MtmOnNodeDisconnect(node+1);
 }
 
 static int MtmWaitSocket(int sd, bool forWrite, timestamp_t timeoutMsec)
@@ -316,22 +316,25 @@ static void MtmCheckResponse(MtmArbiterMessage* resp)
 	} else { 
 		BIT_CLEAR(Mtm->currentLockNodeMask, resp->node-1);
 	}
-
-	// if (BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1))
-	// {
-	// 	MtmStateProcessEvent(MTM_REMOTE_DISABLE);
-	// }
-
-	if (BIT_CHECK(Mtm->disabledNodeMask, resp->node-1) &&
-		sockets[resp->node-1] < 0)
-	{
-		/* We've received heartbeat from disabled node.
+	if (
+		( BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1) || Mtm->status == MTM_IN_MINORITY )
+		&& !BIT_CHECK(Mtm->disabledNodeMask, resp->node-1)
+		&& Mtm->status != MTM_RECOVERY
+		&& Mtm->status != MTM_RECOVERED
+		&& Mtm->nodes[MtmNodeId-1].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) < MtmGetSystemTime()) 
+	{ 
+		MTM_ELOG(WARNING, "Node %d thinks that I'm dead, while I'm %s (message %s)", resp->node, MtmNodeStatusMnem[Mtm->status], MtmMessageKindMnem[resp->code]);
+		BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
+		Mtm->nConfigChanges += 1;
+		MtmSwitchClusterMode(MTM_RECOVERY);
+	} else if (BIT_CHECK(Mtm->disabledNodeMask, resp->node-1) && sockets[resp->node-1] < 0) { 
+		/* We receive heartbeat from disabled node.
 		 * Looks like it is restarted.
 		 * Try to reconnect to it.
 		 */
 		MTM_ELOG(WARNING, "Receive heartbeat from disabled node %d", resp->node);		
 		BIT_SET(Mtm->reconnectMask, resp->node-1);
-	}
+	}	
 }
 
 static void MtmScheduleHeartbeat()
@@ -540,9 +543,17 @@ static void MtmOpenConnections()
 	for (i = 0; i < nNodes; i++) {
 		if (i+1 != MtmNodeId && i < Mtm->nAllNodes) { 
 			sockets[i] = MtmConnectSocket(i, Mtm->nodes[i].con.arbiterPort);
+			if (sockets[i] < 0) { 
+				MtmOnNodeDisconnect(i+1);
+			} 
 		}
 	}
-	MtmStateProcessEvent(MTM_ARBITER_RECEIVER_START);
+	if (Mtm->nLiveNodes < Mtm->nAllNodes/2+1) { /* no quorum */
+		MTM_ELOG(WARNING, "Node is out of quorum: only %d nodes of %d are accessible", Mtm->nLiveNodes, Mtm->nAllNodes);
+		MtmSwitchClusterMode(MTM_IN_MINORITY);
+	} else if (Mtm->status == MTM_INITIALIZATION) { 
+		MtmSwitchClusterMode(MTM_CONNECTED);
+	}
 }
 
 
@@ -575,6 +586,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 			}
 			sockets[node] = MtmConnectSocket(node, Mtm->nodes[node].con.arbiterPort);
 			if (sockets[node] < 0) { 
+				MtmOnNodeDisconnect(node+1);
 				result = false;
 				break;
 			}
@@ -704,18 +716,16 @@ static void MtmSender(Datum arg)
 {
 	int nNodes = MtmMaxNodes;
 	int i;
-	MtmBuffer* txBuffer;
 
 	MtmBackgroundWorker = true;
 
-	txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
+	MtmBuffer* txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
 	MTM_ELOG(LOG, "Start arbiter sender %d", MyProcPid);
 	InitializeTimeouts();
 
 	pqsignal(SIGINT, SetStop);
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
-	pqsignal(SIGHUP, PostgresSigHupHandler);
 
 	/* We're now ready to receive signals */
 	BackgroundWorkerUnblockSignals();
@@ -734,12 +744,6 @@ static void MtmSender(Datum arg)
 		PGSemaphoreLock(&Mtm->sendSemaphore);
 		CHECK_FOR_INTERRUPTS();
 
-		if (ConfigReloadPending)
-		{
-			ConfigReloadPending = false;
-			ProcessConfigFile(PGC_SIGHUP);
-		}
-
 		MtmCheckHeartbeat();
 		/* 
 		 * Use shared lock to improve locality,
@@ -801,7 +805,6 @@ static void MtmMonitor(Datum arg)
 	pqsignal(SIGINT, SetStop);
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
-	pqsignal(SIGHUP, PostgresSigHupHandler);
 
 	MtmBackgroundWorker = true;
 
@@ -816,13 +819,6 @@ static void MtmMonitor(Datum arg)
 		if (rc & WL_POSTMASTER_DEATH) { 
 			break;
 		}
-
-		if (ConfigReloadPending)
-		{
-			ConfigReloadPending = false;
-			ProcessConfigFile(PGC_SIGHUP);
-		}
-
 		MtmRefreshClusterStatus();
 	}
 }
@@ -848,7 +844,6 @@ static void MtmReceiver(Datum arg)
 	pqsignal(SIGINT, SetStop);
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
-	pqsignal(SIGHUP, PostgresSigHupHandler);
 
 	MtmBackgroundWorker = true;
 
@@ -884,14 +879,7 @@ static void MtmReceiver(Datum arg)
 		for (j = 0; j < n; j++) {
 			if (events[j].events & EPOLLIN)  
 #else
-		fd_set events;
-
-		if (ConfigReloadPending)
-		{
-			ConfigReloadPending = false;
-			ProcessConfigFile(PGC_SIGHUP);
-		}
-
+        fd_set events;
 		do { 
 			struct timeval tv;
 			events = inset;
@@ -1018,7 +1006,7 @@ static void MtmReceiver(Datum arg)
 					  default:
 						break;
 					}
-					if (BIT_CHECK(msg->disabledNodeMask, node-1) || BIT_CHECK(Mtm->disabledNodeMask, node-1)) {
+					if (BIT_CHECK(msg->disabledNodeMask, node-1)) {
 						MTM_ELOG(WARNING, "Ignore message from dead node %d\n", node);
 						continue;
 					}
@@ -1096,7 +1084,7 @@ static void MtmReceiver(Datum arg)
 							if (ts->status != TRANSACTION_STATUS_ABORTED) { 
 								MTM_LOG1("Arbiter receive abort message for transaction %s (%llu) from node %d", ts->gid, (long64)ts->xid, node);
 								Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-								ts->abortedByNode = node;
+								ts->aborted_by_node = node;
 								MtmAbortTransaction(ts);
 							}
 							if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
@@ -1173,3 +1161,4 @@ static void MtmReceiver(Datum arg)
 	}
 	proc_exit(1); /* force restart of this bgwroker */
 }
+

contrib/mmts/bgwpool.c

@@ -14,7 +14,6 @@
 
 #include "bgwpool.h"
 #include "multimaster.h"
-#include "utils/guc.h"
 
 bool MtmIsLogicalReceiver;
 int  MtmMaxWorkers;
@@ -44,21 +43,14 @@ static void BgwPoolMainLoop(BgwPool* pool)
 	pqsignal(SIGINT, BgwShutdownWorker);
 	pqsignal(SIGQUIT, BgwShutdownWorker);
 	pqsignal(SIGTERM, BgwShutdownWorker);
-	pqsignal(SIGHUP, PostgresSigHupHandler);
 
     BackgroundWorkerUnblockSignals();
 	BackgroundWorkerInitializeConnection(pool->dbname, pool->dbuser);
 	ActivePortal = &fakePortal;
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";
 
-	while (true) {
-		if (ConfigReloadPending)
-		{
-			ConfigReloadPending = false;
-			ProcessConfigFile(PGC_SIGHUP);
-		}
-
+    while (true) { 
         PGSemaphoreLock(&pool->available);
         SpinLockAcquire(&pool->lock);
 		if (pool->shutdown) {