decouple raftable

Author: kelvich

Cluster.pm

@@ -40,10 +40,10 @@ sub new
 	foreach my $i (1..$nodenum)
 	{
 		my $host = "127.0.0.1";
-		my ($pgport, $raftport) = allocate_ports($host, 2);
+		my ($pgport, $arbiter_port) = allocate_ports($host, 2);
 		my $node = new PostgresNode("node$i", $host, $pgport);
 		$node->{id} = $i;
-		$node->{raftport} = $raftport;
+		$node->{arbiter_port} = $arbiter_port;
 		push(@$nodes, $node);
 	}
 
@@ -71,16 +71,16 @@ sub configure
 {
 	my ($self) = @_;
 	my $nodes = $self->{nodes};
+	my $nnodes = scalar @{ $nodes };
 
-	my $connstr = join(',', map { "${ \$_->connstr('postgres') }" } @$nodes);
-	my $raftpeers = join(',', map { join(':', $_->{id}, $_->host, $_->{raftport}) } @$nodes);
+	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 $raftport = $node->{raftport};
+		my $arbiter_port = $node->{arbiter_port};
 
 		$node->append_conf("postgresql.conf", qq(
 			log_statement = none
@@ -94,20 +94,22 @@ sub configure
 			fsync = off	
 			max_wal_senders = 10
 			wal_sender_timeout = 0
-            default_transaction_isolation = 'repeatable read'
+			default_transaction_isolation = 'repeatable read'
 			max_replication_slots = 10
-			shared_preload_libraries = 'raftable,multimaster'
+			shared_preload_libraries = 'multimaster'
+
+			multimaster.arbiter_port = $arbiter_port
 			multimaster.workers = 10
 			multimaster.queue_size = 10485760 # 10mb
 			multimaster.node_id = $id
 			multimaster.conn_strings = '$connstr'
-			multimaster.use_raftable = false
 			multimaster.heartbeat_recv_timeout = 1000
 			multimaster.heartbeat_send_timeout = 250
-			multimaster.max_nodes = 3
+			multimaster.max_nodes = $nnodes
 			multimaster.ignore_tables_without_pk = true
-			multimaster.twopc_min_timeout = 2000
-            log_line_prefix = '%t: '
+			multimaster.twopc_min_timeout = 50000
+			multimaster.min_2pc_timeout = 50000
+			log_line_prefix = '%t: '
 		));
 
 		$node->append_conf("pg_hba.conf", qq(

Makefile

@@ -1,14 +1,12 @@
 MODULE_big = multimaster
-OBJS = multimaster.o raftable.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
 
 ifndef RAFTABLE_PATH
 RAFTABLE_PATH = ../raftable
 endif
 
 override CPPFLAGS += -I$(RAFTABLE_PATH) -I$(RAFTABLE_PATH)/raft/include
 
-EXTRA_INSTALL = contrib/raftable contrib/mmts
-
 EXTENSION = multimaster
 DATA = multimaster--1.0.sql
 

multimaster.c

@@ -72,9 +72,6 @@
 
 #include "multimaster.h"
 #include "ddd.h"
-#include "raftable_wrapper.h"
-#include "raftable.h"
-#include "worker.h"
 
 typedef struct { 
     TransactionId xid;    /* local transaction ID   */
@@ -215,7 +212,6 @@ int   MtmNodes;
 int   MtmNodeId;
 int   MtmReplicationNodeId;
 int   MtmArbiterPort;
-int   MtmRaftablePort;
 int   MtmConnectTimeout;
 int   MtmReconnectTimeout;
 int   MtmNodeDisableDelay;
@@ -225,7 +221,6 @@ int   MtmHeartbeatSendTimeout;
 int   MtmHeartbeatRecvTimeout;
 int   MtmMin2PCTimeout;
 int   MtmMax2PCRatio;
-bool  MtmUseRaftable;
 bool  MtmUseDtm;
 bool  MtmPreserveCommitOrder;
 bool  MtmVolksWagenMode;
@@ -1745,7 +1740,7 @@ MtmBuildConnectivityMatrix(nodemask_t* matrix, bool nowait)
 
 	for (i = 0; i < n; i++) { 
 		if (i+1 != MtmNodeId) { 
-			void* data = RaftableGet(psprintf("node-mask-%d", i+1), NULL, NULL, nowait);
+			void *data = &Mtm->nodes[i].connectivityMask;
 			if (data == NULL) { 
 				return false;
 			}
@@ -1901,17 +1896,6 @@ void MtmOnNodeDisconnect(int nodeId)
 	MTM_LOG1("Disconnect node %d connectivity mask %llx", nodeId, (long long) Mtm->connectivityMask);
 	MtmUnlock();
 
-	if (!RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false))
-	{
-		elog(WARNING, "Disable node which is in minority according to RAFT");
-		MtmLock(LW_EXCLUSIVE);
-		if (Mtm->status == MTM_ONLINE) { 
-			MtmSwitchClusterMode(MTM_IN_MINORITY);
-		}
-		MtmUnlock();
-		return;
-	}
-
 	MtmSleep(MSEC_TO_USEC(MtmHeartbeatSendTimeout));
 	MtmRefreshClusterStatus(false);
 }
@@ -1922,9 +1906,6 @@ void MtmOnNodeConnect(int nodeId)
 	BIT_CLEAR(Mtm->connectivityMask, nodeId-1);
 	BIT_CLEAR(Mtm->reconnectMask, nodeId-1);
 	MtmUnlock();
-
-	MTM_LOG1("Reconnect node %d, connectivityMask=%llx", nodeId, (long long) Mtm->connectivityMask);
-	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false); 
 }
 
 
@@ -2034,21 +2015,6 @@ static void MtmLoadLocalTables(void)
 		heap_close(rel, RowExclusiveLock);
 	}
 }
-	
-static void MtmRaftableInitialize()
-{
-	int i;
-
-	for (i = 0; i < MtmNodes; i++)
-	{
-		int port = MtmConnections[i].raftablePort;
-		if (port == 0) {
-			port = MtmRaftablePort + i;
-		}
-		raftable_peer(i, MtmConnections[i].hostName, port);
-	}
-	raftable_start(MtmNodeId - 1);
-}
 
 static void MtmCheckControlFile(void)
 {
@@ -2197,19 +2163,10 @@ void MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* connStr)
 	memcpy(conn->hostName, host, hostLen);
 	conn->hostName[hostLen] = '\0';
 
-	port = strstr(connStr, "raftport=");
+	port = strstr(connStr, "arbiter_port=");
 	if (port != NULL) {
-		if (sscanf(port+9, "%d", &conn->raftablePort) != 1) { 
-			elog(ERROR, "Invalid raftable port: %s", port+9);
-		}
-	} else { 
-		conn->raftablePort = 0;
-	}
-
-	port = strstr(connStr, "arbiterport=");
-	if (port != NULL) {
-		if (sscanf(port+12, "%d", &conn->arbiterPort) != 1) { 
-			elog(ERROR, "Invalid arbiter port: %s", port+12);
+		if (sscanf(port+13, "%d", &conn->arbiterPort) != 1) {
+			elog(ERROR, "Invalid arbiter port: %s", port+13);
 		}
 	} else { 
 		conn->arbiterPort = MULTIMASTER_DEFAULT_ARBITER_PORT;
@@ -2538,19 +2495,6 @@ _PG_init(void)
 		NULL
 	);
 
-	DefineCustomBoolVariable(
-		"multimaster.use_raftable",
-		"Use raftable plugin for internode communication",
-		NULL,
-		&MtmUseRaftable,
-		true,
-		PGC_BACKEND,
-		0,
-		NULL,
-		NULL,
-		NULL
-	);
-
 	DefineCustomBoolVariable(
 		"multimaster.ignore_tables_without_pk",
 		"Do not replicate tables withpout primary key",
@@ -2708,21 +2652,6 @@ _PG_init(void)
 		NULL
 	);
 
-	DefineCustomIntVariable(
-		"multimaster.raftable_port",
-		"Base value for assigning raftable ports",
-		NULL,
-		&MtmRaftablePort,
-		6543,
-	    0,
-		INT_MAX,
-		PGC_BACKEND,
-		0,
-		NULL,
-		NULL,
-		NULL
-	);
-
 	DefineCustomStringVariable(
 		"multimaster.conn_strings",
 		"Multimaster node connection strings separated by commas, i.e. 'replication=database dbname=postgres host=localhost port=5001,replication=database dbname=postgres host=localhost port=5002'",
@@ -2813,9 +2742,6 @@ _PG_init(void)
 
     BgwPoolStart(MtmWorkers, MtmPoolConstructor);
 
-	if (MtmUseRaftable) {
-		MtmRaftableInitialize();
-	}
 	MtmArbiterInitialize();
 
 	/*
@@ -3516,8 +3442,7 @@ PGconn *PQconnectdb_safe(const char *conninfo)
 {
 	PGconn *conn;
 	char *safe_connstr = pstrdup(conninfo);
-	erase_option_from_connstr("raftport", safe_connstr);
-	erase_option_from_connstr("arbiterport", safe_connstr);
+	erase_option_from_connstr("arbiter_port", safe_connstr);
 
 	conn = PQconnectdb(safe_connstr);
 
@@ -3626,11 +3551,17 @@ Datum mtm_dump_lock_graph(PG_FUNCTION_ARGS)
 	int i;
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
-		size_t size;
-		char* data = RaftableGet(psprintf("lock-graph-%d", i+1), &size, NULL, false);
-		if (data) { 
-			GlobalTransactionId *gtid = (GlobalTransactionId *)data;
-			GlobalTransactionId *last = (GlobalTransactionId *)(data + size);
+		size_t lockGraphSize;
+		char  *lockGraphData;
+		MtmLockNode(i + MtmMaxNodes, LW_SHARED);
+		lockGraphSize = Mtm->nodes[i].lockGraphUsed;
+		lockGraphData = palloc(lockGraphSize);
+		memcpy(lockGraphData, Mtm->nodes[i].lockGraphData, lockGraphSize);
+		MtmUnlockNode(i + MtmMaxNodes);
+
+		if (lockGraphData) {
+			GlobalTransactionId *gtid = (GlobalTransactionId *) lockGraphData;
+			GlobalTransactionId *last = (GlobalTransactionId *) (lockGraphData + lockGraphSize);
 			appendStringInfo(s, "node-%d lock graph: ", i+1);
 			while (gtid != last) { 
 				GlobalTransactionId *src = gtid++;
@@ -4543,19 +4474,28 @@ MtmDetectGlobalDeadLockFortXid(TransactionId xid)
 
         ByteBufferAlloc(&buf);
         EnumerateLocks(MtmSerializeLock, &buf);
-		RaftableSet(psprintf("lock-graph-%d", MtmNodeId), buf.data, buf.used, false);
+
+		Assert(replorigin_session_origin == InvalidRepOriginId);
+		XLogFlush(LogLogicalMessage("L", buf.data, buf.used, false));
+
 		MtmSleep(MSEC_TO_USEC(DeadlockTimeout));
 		MtmGraphInit(&graph);
 		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.used/sizeof(GlobalTransactionId));
         ByteBufferFree(&buf);
 		for (i = 0; i < Mtm->nAllNodes; i++) { 
 			if (i+1 != MtmNodeId && !BIT_CHECK(Mtm->disabledNodeMask, i)) { 
-				size_t size;
-				void* data = RaftableGet(psprintf("lock-graph-%d", i+1), &size, NULL, false);
-				if (data == NULL) { 
+				size_t lockGraphSize;
+				void* lockGraphData;
+				MtmLockNode(i + MtmMaxNodes, LW_SHARED);
+				lockGraphSize = Mtm->nodes[i].lockGraphUsed;
+				lockGraphData = palloc(lockGraphSize);
+				memcpy(lockGraphData, Mtm->nodes[i].lockGraphData, lockGraphSize);
+				MtmUnlockNode(i + MtmMaxNodes);
+
+				if (lockGraphData == NULL) {
 					return true; /* If using Raftable is disabled */
 				} else { 
-					MtmGraphAdd(&graph, (GlobalTransactionId*)data, size/sizeof(GlobalTransactionId));
+					MtmGraphAdd(&graph, (GlobalTransactionId*)lockGraphData, lockGraphSize/sizeof(GlobalTransactionId));
 				}
 			}
 		}