Fix issues with deadlock detection

Author: knizhnik

arbiter.c

@@ -335,7 +335,7 @@ static void MtmOpenConnections()
 		if (i+1 != MtmNodeId) { 
 			sockets[i] = MtmConnectSocket(host, MtmArbiterPort + i + 1, MtmConnectAttempts);
 			if (sockets[i] < 0) { 
-				MtmOnLostConnection(i+1);
+				MtmOnNodeDisconnect(i+1);
 			} 
 		} else {
 			sockets[i] = -1;
@@ -360,7 +360,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 		}
 		sockets[node] = MtmConnectSocket(hosts[node], MtmArbiterPort + node + 1, MtmReconnectAttempts);
 		if (sockets[node] < 0) { 
-			MtmOnLostConnection(node+1);
+			MtmOnNodeDisconnect(node+1);
 			return false;
 		}
 	}
@@ -403,7 +403,7 @@ static void MtmAcceptOneConnection()
 			} else { 
 				elog(NOTICE, "Arbiter established connection with node %d", msg.node); 
 				BIT_CLEAR(ds->connectivityMask, msg.node-1);
-				MtmOnConnectNode(msg.node);
+				MtmOnNodeConnect(msg.node);
 				MtmRegisterSocket(fd, msg.node-1);
 				sockets[msg.node-1] = fd;
 			}

bkb.c

@@ -1,4 +1,4 @@
-#include "postgres.h"
+#include <stdint.h>
 #include "bkb.h"
 
 /*
@@ -12,7 +12,6 @@ typedef struct {
 
 static void list_append(NodeList* list, int n)
 {
-	Assert(list->size < MAX_NODES);
 	list->nodes[list->size++] = n;
 }
 
@@ -44,7 +43,7 @@ static void findMaximumIndependentSet(NodeList* cur, NodeList* result, nodemask_
 		int pos = -1;
 
 		for (j = ne; j < ce; j++) { 
-			if (!BIT_CHECK(graph[p], oldSet[j])) {
+			if (BIT_CHECK(graph[p], oldSet[j])) {
 				if (++cnt == minnod) { 
 					break;
 				}
@@ -71,13 +70,13 @@ static void findMaximumIndependentSet(NodeList* cur, NodeList* result, nodemask_
 
 		newne = 0;
 		for (i = 0; i < ne; i++) {
-			if (BIT_CHECK(graph[sel], oldSet[i])) {
+			if (!BIT_CHECK(graph[sel], oldSet[i])) {
 				newSet[newne++] = oldSet[i];
 			}
 		}
 	    newce = newne;
 		for (i = ne + 1; i < ce; i++) {
-			if (BIT_CHECK(graph[sel], oldSet[i])) { 
+			if (!BIT_CHECK(graph[sel], oldSet[i])) { 
 				newSet[newce++] = oldSet[i];
 			}
 		}
@@ -92,8 +91,9 @@ static void findMaximumIndependentSet(NodeList* cur, NodeList* result, nodemask_
 			}
 		}
 		cur->size -= 1;
+		ne += 1;
 		if (k > 1) {
-			for (s = ++ne; BIT_CHECK(graph[fixp], oldSet[s]); s++);
+			for (s = ++ne; !BIT_CHECK(graph[fixp], oldSet[s]); s++);
 		}
 	}
 }
@@ -114,7 +114,7 @@ nodemask_t MtmFindMaxClique(nodemask_t* graph, int n_nodes)
 	findMaximumIndependentSet(&tmp, &result, graph, all, 0, n_nodes);
 	mask = 0;
 	for (i = 0; i < result.size; i++) { 
-		mask |= (nodemask_t)1 << result.nodes[i];
+		BIT_SET(mask, result.nodes[i]);
 	}
 	return mask;
 }

bkbtest.c

@@ -0,0 +1,16 @@
+#include <stdio.h>
+#include <stdint.h>
+#include "bkb.h"
+
+int main() { 
+	nodemask_t matrix[64] = {0};
+	nodemask_t clique;
+	matrix[0] = 6;
+	matrix[1] = 4;
+	matrix[2] = 1;
+	matrix[4] = 3;
+	clique = MtmFindMaxClique(matrix, 64);
+	printf("Clique=%lx\n", clique);
+	return 0;
+}
+	

multimaster.c

@@ -141,6 +141,7 @@ int   MtmArbiterPort;
 int   MtmNodes;
 int   MtmConnectAttempts;
 int   MtmConnectTimeout;
+int   MtmKeepaliveTimeout;
 int   MtmReconnectAttempts;
 
 static int MtmQueueSize;
@@ -986,6 +987,21 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomIntVariable(
+		"multimaster.keepalive_timeout",
+		"Multimaster keepalive interval for sockets",
+		"Timeout in microseconds before polling state of nodes",
+		&MtmKeepaliveTimeout,
+		1000000,
+		1,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.connect_attempts",
 		"Multimaster number of connect attemts",
@@ -1528,8 +1544,12 @@ MtmGetGtid(TransactionId xid, GlobalTransactionId* gtid)
 
 	MtmLock(LW_SHARED);
 	ts = (MtmTransState*)hash_search(xid2state, &xid, HASH_FIND, NULL);
-	Assert(ts != NULL);
-	*gtid = ts->gtid;
+	if (ts != NULL) { 
+		*gtid = ts->gtid;
+	} else { 
+		gtid->node = MtmNodeId;
+		gtid->xid = xid;
+	}
 	MtmUnlock();
 }
 
@@ -1601,15 +1621,19 @@ MtmDetectGlobalDeadLock(PGPROC* proc)
 
         ByteBufferAlloc(&buf);
         EnumerateLocks(MtmSerializeLock, &buf);
-        ByteBufferFree(&buf);
-		PaxosSet(psprintf("lock-graph-%d", MtmNodeId), buf.data, buf.size);
+		PaxosSet(psprintf("lock-graph-%d", MtmNodeId), buf.data, buf.used);
 		MtmGraphInit(&graph);
-		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.size/sizeof(GlobalTransactionId));
+		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.used/sizeof(GlobalTransactionId));
+        ByteBufferFree(&buf);
 		for (i = 0; i < MtmNodes; i++) { 
 			if (i+1 != MtmNodeId && !BIT_CHECK(dtm->disabledNodeMask, i)) { 
 				int size;
 				void* data = PaxosGet(psprintf("lock-graph-%d", i+1), &size, NULL);
-				MtmGraphAdd(&graph, (GlobalTransactionId*)data, size/sizeof(GlobalTransactionId));
+				if (data == NULL) { 
+					hasDeadlock = true; /* Just temporary hack until no Paxos */
+				} else { 
+					MtmGraphAdd(&graph, (GlobalTransactionId*)data, size/sizeof(GlobalTransactionId));
+				}
 			}
 		}
 		MtmGetGtid(pgxact->xid, &gtid);
@@ -1636,42 +1660,56 @@ MtmBuildConnectivityMatrix(nodemask_t* matrix)
 
 void MtmUpdateClusterStatus(void)
 {
-	nodemask_t mask, clique;
+	nodemask_t mask, clique, disconnectedMask;
 	nodemask_t matrix[MAX_NODES];
 	int i;
 
 	MtmBuildConnectivityMatrix(matrix);
 
 	clique = MtmFindMaxClique(matrix, MtmNodes);
-
+	disconnectedMask = ~clique & (((nodemask_t)1 << MtmNodes)-1);
 	MtmLock(LW_EXCLUSIVE);
-	mask = clique & ~dtm->disabledNodeMask;
+	mask = disconnectedMask & ~dtm->disabledNodeMask;
 	for (i = 0; mask != 0; i++, mask >>= 1) {
 		if (mask & 1) { 
 			dtm->nNodes -= 1;
 			BIT_SET(dtm->disabledNodeMask, i);
 		}
 	}
-	if (dtm->disabledNodeMask != clique) { 
-		dtm->disabledNodeMask |= clique;
+	if (dtm->disabledNodeMask != disconnectedMask) { 
+		dtm->disabledNodeMask |= disconnectedMask;
 		PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->disabledNodeMask, sizeof dtm->disabledNodeMask);
 	}
 	MtmUnlock();
 }
 
-void MtmOnLostConnection(int nodeId)
+void MtmOnNodeDisconnect(int nodeId)
 {
 	BIT_SET(dtm->connectivityMask, nodeId-1);
 	PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->connectivityMask, sizeof dtm->connectivityMask);
 
 	/* Wait more than socket KEEPALIVE timeout to let other nodes update their statuses */
-	MtmSleep(MtmConnectTimeout);
+	MtmSleep(MtmKeepaliveTimeout);
 
 	MtmUpdateClusterStatus();
 }
 
-void MtmOnConnectNode(int nodeId)
+void MtmOnNodeConnect(int nodeId)
 {
 	BIT_CLEAR(dtm->connectivityMask, nodeId-1);
 	PaxosSet(psprintf("node-mask-%d", MtmNodeId), &dtm->connectivityMask, sizeof dtm->connectivityMask);
 }
+
+/*
+ * Paxos function stubs (until them are miplemented)
+ */
+void* PaxosGet(char const* key, int* size, PaxosTimestamp* ts)
+{
+	if (size != NULL) { 
+		*size = 0;
+	}
+	return NULL;
+}
+
+void  PaxosSet(char const* key, void const* value, int size)
+{}

multimaster.h

@@ -141,8 +141,8 @@ extern void  MtmAdjustSubtransactions(MtmTransState* ts);
 extern void  MtmLock(LWLockMode mode);
 extern void  MtmUnlock(void);
 extern void  MtmDropNode(int nodeId, bool dropSlot);
-extern void  MtmOnLostConnection(int nodeId);
-extern void  MtmOnConnectNode(int nodeId);
+extern void  MtmOnNodeDisconnect(int nodeId);
+extern void  MtmOnNodeConnect(int nodeId);
 extern MtmState* MtmGetState(void);
 extern timestamp_t MtmGetCurrentTime(void);
 extern void  MtmSleep(timestamp_t interval);

tests/dtmbench.cpp

@@ -195,7 +195,7 @@ void initializeDatabase()
 		exec(txn, "insert into t (select generate_series(0,%d), %d)", cfg.nAccounts-1, 0);
 		txn.commit();
 	}
-	printf("Initialization completed in %f seconds\n", (start  - getCurrentTime())/100000.0);
+	printf("Initialization completed in %f seconds\n", (getCurrentTime() - start)/100000.0);
 }
 
 int main (int argc, char* argv[])