Revert "Revert "Merge branch 'PGPROEE9_6' into PGPROEE9_6_MULTIMASTER""

This reverts commit 1460198.

Author: vbwagner

contrib/Makefile

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

contrib/mmts/Cluster.pm

@@ -66,6 +66,7 @@ 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);
 	}
 
@@ -89,47 +90,54 @@ 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 = join(', ', map { "${ \$_->connstr('postgres') } arbiter_port=${ \$_->{arbiter_port} }" } @$nodes);
+	my $connstr = $self->all_connstrs();
 
 	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_socket_directories = '$unix_sock_dir'
 			port = $pgport
 			max_prepared_transactions = 10
 			max_connections = 10
 			max_worker_processes = 100
 			wal_level = logical
-			max_wal_senders = 5
+			max_wal_senders = 6
 			wal_sender_timeout = 0
 			default_transaction_isolation = 'repeatable read'
-			max_replication_slots = 5
+			max_replication_slots = 6
 			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 = 2050
+			multimaster.heartbeat_recv_timeout = 1050
 			multimaster.heartbeat_send_timeout = 250
-			multimaster.max_nodes = $nnodes
-			multimaster.ignore_tables_without_pk = true
+			multimaster.max_nodes = 6
+			multimaster.ignore_tables_without_pk = false
 			multimaster.queue_size = 4194304
-			multimaster.min_2pc_timeout = 150000
 			log_line_prefix = '%t: '
 		));
 

contrib/mmts/Dockerfile

@@ -6,6 +6,9 @@ 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
+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
 MODULE_big = multimaster
 
 PG_CPPFLAGS = -I$(libpq_srcdir)

contrib/mmts/arbiter.c

@@ -76,6 +76,7 @@
 
 
 #include "multimaster.h"
+#include "state.h"
 
 #define MAX_ROUTES       16
 #define INIT_BUFFER_SIZE 1024
@@ -189,7 +190,6 @@ 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,25 +316,22 @@ static void MtmCheckResponse(MtmArbiterMessage* resp)
 	} else { 
 		BIT_CLEAR(Mtm->currentLockNodeMask, resp->node-1);
 	}
-	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.
+
+	// 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.
 		 * 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()
@@ -543,17 +540,9 @@ 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);
-			} 
 		}
 	}
-	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);
-	}
+	MtmStateProcessEvent(MTM_ARBITER_RECEIVER_START);
 }
 
 
@@ -586,7 +575,6 @@ 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;
 			}
@@ -716,16 +704,18 @@ static void MtmSender(Datum arg)
 {
 	int nNodes = MtmMaxNodes;
 	int i;
+	MtmBuffer* txBuffer;
 
 	MtmBackgroundWorker = true;
 
-	MtmBuffer* txBuffer = (MtmBuffer*)palloc0(sizeof(MtmBuffer)*nNodes);
+	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();
@@ -744,6 +734,12 @@ 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,
@@ -805,6 +801,7 @@ static void MtmMonitor(Datum arg)
 	pqsignal(SIGINT, SetStop);
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
+	pqsignal(SIGHUP, PostgresSigHupHandler);
 
 	MtmBackgroundWorker = true;
 
@@ -819,6 +816,13 @@ static void MtmMonitor(Datum arg)
 		if (rc & WL_POSTMASTER_DEATH) { 
 			break;
 		}
+
+		if (ConfigReloadPending)
+		{
+			ConfigReloadPending = false;
+			ProcessConfigFile(PGC_SIGHUP);
+		}
+
 		MtmRefreshClusterStatus();
 	}
 }
@@ -844,6 +848,7 @@ static void MtmReceiver(Datum arg)
 	pqsignal(SIGINT, SetStop);
 	pqsignal(SIGQUIT, SetStop);
 	pqsignal(SIGTERM, SetStop);
+	pqsignal(SIGHUP, PostgresSigHupHandler);
 
 	MtmBackgroundWorker = true;
 
@@ -879,7 +884,14 @@ static void MtmReceiver(Datum arg)
 		for (j = 0; j < n; j++) {
 			if (events[j].events & EPOLLIN)  
 #else
-        fd_set events;
+		fd_set events;
+
+		if (ConfigReloadPending)
+		{
+			ConfigReloadPending = false;
+			ProcessConfigFile(PGC_SIGHUP);
+		}
+
 		do { 
 			struct timeval tv;
 			events = inset;
@@ -1006,7 +1018,7 @@ static void MtmReceiver(Datum arg)
 					  default:
 						break;
 					}
-					if (BIT_CHECK(msg->disabledNodeMask, node-1)) {
+					if (BIT_CHECK(msg->disabledNodeMask, node-1) || BIT_CHECK(Mtm->disabledNodeMask, node-1)) {
 						MTM_ELOG(WARNING, "Ignore message from dead node %d\n", node);
 						continue;
 					}
@@ -1084,7 +1096,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->aborted_by_node = node;
+								ts->abortedByNode = node;
 								MtmAbortTransaction(ts);
 							}
 							if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
@@ -1161,4 +1173,3 @@ static void MtmReceiver(Datum arg)
 	}
 	proc_exit(1); /* force restart of this bgwroker */
 }
-

contrib/mmts/bgwpool.c

@@ -14,6 +14,7 @@
 
 #include "bgwpool.h"
 #include "multimaster.h"
+#include "utils/guc.h"
 
 bool MtmIsLogicalReceiver;
 int  MtmMaxWorkers;
@@ -43,14 +44,21 @@ 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) { 
+	while (true) {
+		if (ConfigReloadPending)
+		{
+			ConfigReloadPending = false;
+			ProcessConfigFile(PGC_SIGHUP);
+		}
+
         PGSemaphoreLock(&pool->available);
         SpinLockAcquire(&pool->lock);
 		if (pool->shutdown) {