Distributed deadlock detections

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) { 
-				MtmDropNode(i+1, false);
+				MtmOnLostConnection(i+1);
 			} 
 		} else {
 			sockets[i] = -1;
@@ -358,7 +358,7 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 		close(sockets[node]);
 		sockets[node] = MtmConnectSocket(hosts[node], MtmArbiterPort + node + 1, MtmReconnectAttempts);
 		if (sockets[node] < 0) { 
-			MtmDropNode(node+1, false);
+			MtmOnLostConnection(node+1);
 			return false;
 		}
 	}

ddd.c

@@ -0,0 +1,74 @@
+#include <stddef.h>
+#include <stdlib.h>
+#include <string.h>
+#include "ddd.h"
+
+
+void MtmGraphInit(MtmGraph* graph)
+{
+    memset(graph->hashtable, 0, sizeof(graph->hashtable));
+}
+
+static inline MtmVertex* findVertex(MtmGraph* graph, GlobalTransactionId* gtid)
+{
+    xid_t h = gtid->xid % MAX_TRANSACTIONS;
+    MtmVertex* v;
+    for (v = graph->hashtable[h]; v != NULL; v = v->next) { 
+        if (v->gtid == *gtid) { 
+            return v;
+        }
+    }
+	v = (MtmVertex*)palloc(sizeof(MtmVertex));
+    v->gtid = *gtid;
+	v->outgoingEdges = NULL;
+    v->collision = graph->hashtable[h];
+    graph->hashtable[h] = v;
+    return v;
+}
+
+void MtmGraphAdd(MtmGraph* graph, GlobalTransactionId* gtid, int size)
+{
+    GlobalTransactionId* last = gtid + size;
+    MtmEdge *e, *next, *edges = NULL;
+    while (gtid != last) { 
+        Vertex* src = findVertex(graph, gtid++);
+        while (gtid->node != 0) { 
+            Vertex* dst = findVertex(graph, gtid++);
+            e = (MtmEdge*)palloc(sizeof(MtmEdge));
+            dst->nIncomingEdges += 1;
+            e->dst = dst;
+            e->src = src;
+            e->next = v->outgoingEdges;
+            v->outgoingEdges = e;
+        }
+		gtid += 1;
+    }
+}
+
+static bool recursiveTraverseGraph(MtmVertex* root, MtmVertex* v)
+{
+    Edge* e;
+    v->node = VISITED_NODE_MARK;
+    for (e = v->outgoingEdges; e != NULL; e = e->next) {
+        if (e->dst == root) { 
+            return true;
+        } else if (e->dst->node != VISITED_NODE_MAR && recursiveTraverseGraph(root, e->dst)) { /* loop */
+            return true;
+        } 
+    }
+    return false;        
+}
+
+bool MtmGraphFindLoop(MtmGraph* graph, GlobalTransactionId* root)
+{
+    Vertex* v;
+    for (v = graph->hashtable[root->xid % MAX_TRANSACTIONS]; v != NULL; v = v->next) { 
+        if (v->gtid == *root) { 
+            if (recursiveTraverseGraph(v, v)) { 
+                return true;
+            }
+            break;
+        }
+    }
+    return false;        
+}

ddd.h

@@ -0,0 +1,30 @@
+#ifndef __DDD_H__
+
+#include "multimaster.h"
+
+#define MAX_TRANSACTIONS  1024
+#define VISITED_NODE_MARK 0
+
+typedef struct MtmEdge {
+	struct MtmEdge*   next; /* list of outgoing edges */
+    struct MtmVertex* dst;
+    struct MtmVertex* src;
+} MtmEdge;
+
+typedef struct MtmVertex
+{    
+    struct MtmEdge* outgoingEdges;
+    struct MtmVertex* collision;
+	GlobalTransactionId gtid;
+} MtmVertex;
+
+typedef struct MtmGraph
+{
+    MtmVertex* hashtable[MTM_MAX_TRANSACTIONS];
+} Graph;
+
+extern void MtmGraphInit(MtmGraph* graph);
+extern void MtmGraphAdd(MtmGraph* graph, GlobalTransactionId* subgraph, int size);
+extern bool MtmGraphFindLoop(MtmGraph* graph, GlobalTransactionId* root);
+
+#endif

multimaster.c

@@ -112,6 +112,7 @@ static BgwPool* MtmPoolConstructor(void);
 static bool MtmRunUtilityStmt(PGconn* conn, char const* sql);
 static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError);
 static void MtmVoteForTransaction(MtmTransState* ts);
+static void MtmSerializeLockGraph(ByteBuffer* buf)
 
 static HTAB* xid2state;
 static MtmCurrentTrans dtmTx;
@@ -805,13 +806,6 @@ MtmSetTransactionStatus(TransactionId xid, int nsubxids, TransactionId *subxids,
 	PgTransactionIdSetTreeStatus(xid, nsubxids, subxids, status, lsn);
 }
 
-static bool 
-MtmDetectGlobalDeadLock(PGPROC* proc)
-{
-    elog(WARNING, "Global deadlock?");
-    return true;
-}
-
 static void 
 MtmShmemStartup(void)
 {
@@ -1582,7 +1576,63 @@ MtmSerializeLock(PROCLOCK* proclock, void* arg)
     }
 }
 
-void MtmSerializeLockGraph(ByteBuffer* buf)
+/* Stubs */
+typedef PaxosTimestamp { 
+	time_t time;   /* local time at master */
+	uint32 naster; /* master node for this operation */
+	uint32 psn;    /* PAXOS serial number */
+} PaxosTimestamp;
+
+void* PaxosGet(char const* key, int* size, PaxosTimestamp* ts);
+void  PaxosSet(char const* key, void* value, int size);
+
+
+static bool 
+MtmDetectGlobalDeadLock(PGPROC* proc)
+{
+    bool hasDeadlock = false;
+    ByteBuffer buf;
+    PGXACT* pgxact = &ProcGlobal->allPgXact[proc->pgprocno];
+	bool hasDeadlock = false;
+    if (TransactionIdIsValid(pgxact->xid)) { 
+		MtmGraph graph;
+		GlobalTransactionId gtid; 
+		int i;
+		
+        ByteBufferAlloc(&buf);
+        EnumerateLocks(DtmSerializeLock, &buf);
+        ByteBufferFree(&buf);
+		PaxosPut(dsprintf("lock-graph-%d", MMNodeId), buf.data, buf.size);
+		MtmGraphInit(&graph);
+		MtmGraphAdd(&graph, (GlobalTransactionId*)buf.data, buf.size/sizeof(GlobalTransactionId));
+		for (i = 0; i < MtmNodes; i++) { 
+			if (i+1 MtmNodeId && !BIT_CHECK(dtm->disabledNodeMask, i)) { 
+				int size;
+				void* data = PaxosGet(dsprintf("lock-graph-%d", i+1), &size, NULL);
+				MtmGraphAdd(&graph, (GlobalTransactionId*)data, size/sizeof(GlobalTransactoinId));
+			}
+		}
+		MtmGetGtid(pgxact->xid, &gtid);
+		hasDeadlock = MtmGraphFindLoop(&graph, &gtid);
+		elog(WARNING, "Distributed deadlock check for %u:%u = %d", gtid.node, gtid.xid, hasDeadlock);
+	}
+    return hasDeadlock;
+}
+
+void MtmOnLostConnection(int nodeId)
 {
-	EnumerateLocks(MtmSerializeLock, buf);
+	int i;
+	nodemask_t mask = dtm->disabledNodeMask;
+	nodemask_t matrix = (nodemask_t*)palloc0(sizeof(nodemask_t)*MtmNodes);
+	
+	mask |= (nodemask_t)1 << (nodeId-1);
+	PaxosPut(dsprintf("node-mask-%d", MMNodeId), &mask, sizeof mask);
+	matrix[MtmNodeId-1] = mask;
+	MtmSleep(MtmConnectTimeout);
+	for (i = 0; i < MtnNodes; i++) { 
+		if (i+1 != MtmNodeId && !BIT_CHECK(dtm->disabledNodeMask, i)) { 
+			void* data = PaxosGet(dsprintf("node-mask-%d", i+1), NULL, NULL);
+			matrix[i] = *(nodemask_t*)data;
+		}
+	}
 }

multimaster.h

@@ -141,10 +141,10 @@ 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 MtmState* MtmGetState(void);
 extern timestamp_t MtmGetCurrentTime(void);
 extern void  MtmSleep(timestamp_t interval);
 extern bool  MtmIsRecoveredNode(int nodeId);
-extern void  MtmSerializeLockGraph(ByteBuffer* buf);
 
 #endif