Merge branch 'PGPROEE9_6_MULTIMASTER' of gitlab.postgrespro.ru:pgpro-dev/postgrespro into PGPROEE9_6_MULTIMASTER

Author: vbwagner

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
+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

@@ -467,8 +467,12 @@ static int MtmConnectSocket(int node, int port)
 				MTM_ELOG(WARNING, "Arbiter trying to connect to %s:%d: %s", host, port, strerror(errcode));
 				goto Error;
 			}
-		} else {
-			MTM_ELOG(WARNING, "Arbiter waiting socket to %s:%d: %s", host, port, strerror(errno));
+		} else if (rc == 0) {
+			MTM_ELOG(WARNING, "Arbiter failed to connect to socket to %s:%d within specified timeout", host, port);
+			goto Error;
+		} else { 
+			MTM_ELOG(WARNING, "Arbiter failed to wait socket to %s:%d: %s", host, port, strerror(errno));
+			goto Error;
 		}
 	}
 	else if (rc != 0) {

contrib/mmts/bkb.h

@@ -14,6 +14,7 @@ typedef ulong64 nodemask_t;
 #define BIT_CHECK(mask, bit) (((mask) & ((nodemask_t)1 << (bit))) != 0)
 #define BIT_CLEAR(mask, bit) (mask &= ~((nodemask_t)1 << (bit)))
 #define BIT_SET(mask, bit)   (mask |= ((nodemask_t)1 << (bit)))
+#define ALL_BITS ((nodemask_t)~0)
 
 extern nodemask_t MtmFindMaxClique(nodemask_t* matrix, int n_modes, int* clique_size);
 

contrib/mmts/multimaster--1.0.sql

@@ -51,7 +51,7 @@ AS 'MODULE_PATHNAME','mtm_get_nodes_state'
 LANGUAGE C;
 
 CREATE TYPE mtm.cluster_state AS ("id" integer, "status" text, "disabledNodeMask" bigint, "disconnectedNodeMask" bigint, "catchUpNodeMask" bigint, "liveNodes" integer, "allNodes" integer, "nActiveQueries" integer, "nPendingQueries" integer, "queueSize" bigint, "transCount" bigint, "timeShift" bigint, "recoverySlot" integer,
-"xidHashSize" bigint, "gidHashSize" bigint, "oldestXid" bigint, "configChanges" integer, "stalledNodeMask" bigint, "stoppedNodeMask" bigint, "lastStatusChange" timestamp);
+"xidHashSize" bigint, "gidHashSize" bigint, "oldestXid" bigint, "configChanges" integer, "stalledNodeMask" bigint, "stoppedNodeMask" bigint, "deadNodeMask" bigint, "lastStatusChange" timestamp);
 
 CREATE TYPE mtm.trans_state AS ("status" text, "gid" text, "xid" bigint, "coordinator" integer, "gxid" bigint, "csn" timestamp, "snapshot" timestamp, "local" boolean, "prepared" boolean, "active" boolean, "twophase" boolean, "votingCompleted" boolean, "participants" bigint, "voted" bigint, "configChanges" integer);
 
@@ -75,6 +75,14 @@ CREATE FUNCTION mtm.make_table_local(relation regclass) RETURNS void
 AS 'MODULE_PATHNAME','mtm_make_table_local'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.broadcast_table(srcTable regclass, dstNodesMask bigint) RETURNS void
+AS 'MODULE_PATHNAME','mtm_broadcast_table'
+LANGUAGE C;
+
+CREATE FUNCTION mtm.copy_table(srcTable regclass, dstNode integer) RETURNS void
+AS 'MODULE_PATHNAME','mtm_copy_table'
+LANGUAGE C;
+
 CREATE FUNCTION mtm.dump_lock_graph() RETURNS text
 AS 'MODULE_PATHNAME','mtm_dump_lock_graph'
 LANGUAGE C;
@@ -91,6 +99,10 @@ CREATE FUNCTION mtm.check_deadlock(xid bigint) RETURNS boolean
 AS 'MODULE_PATHNAME','mtm_check_deadlock'
 LANGUAGE C;
 
+CREATE FUNCTION mtm.referee_poll(xid bigint) RETURNS bigint
+AS 'MODULE_PATHNAME','mtm_referee_poll'
+LANGUAGE C;
+
 CREATE TABLE IF NOT EXISTS mtm.local_tables(rel_schema text, rel_name text, primary key(rel_schema, rel_name));
 
 CREATE OR REPLACE FUNCTION mtm.alter_sequences() RETURNS boolean AS