Add ddd.c to dtmd

Author: knizhnik

contrib/pg_dtm/docs/DTM.odp

@@ -15,11 +15,11 @@ all: bin/dtmd
 	@echo Done.
 	@echo Feel free to run the tests with \'make check\'.
 
-bin/dtmd: obj/server.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o | bindir objdir
+bin/dtmd: obj/server.o obj/main.o obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o obj/snapshot.o obj/ddd.o | bindir objdir
 	$(CC) -o bin/dtmd $(CFLAGS) \
 		obj/server.o obj/main.o \
 		obj/clog.o obj/clogfile.o obj/util.o obj/transaction.o \
-		obj/snapshot.o \
+		obj/snapshot.o obj/ddd.o \
 		$(SOCKHUB_LDFLAGS)
 
 obj/server.o: src/server.c | objdir

contrib/pg_dtm/docs/dtm.pdf

@@ -34,7 +34,7 @@ typedef struct Transaction {
 	void *listeners[CHAR_TO_INDEX('z')]; // we are going to use 'a' to 'z' for indexing
 } Transaction;
 
-static inline void l2_list_is_empty(L2List* elem)
+static inline bool l2_list_is_empty(L2List* elem)
 {
     return elem->next == elem;
 }

contrib/pg_dtm/dtmd/Makefile

@@ -1,39 +1,19 @@
-#include "transaction.h"
+#include <stddef.h>
+#include <stdlib.h>
+#include <string.h>
+#include "ddd.h"
 
-typedef struct Instance {
-    struct Edge* edges; /* local subgraph */    
-} Instance;
-
-typedef struct Edge {
-    L2List node; /* node of list of outgoing eedges */
-    struct Edge* next;  /* list of edges of local subgraph */  
-    struct Vertex* dst;
-    struct Vertex* src;
-} Edge;
-
-typedef struct Vertex
-{
-    L2List outgoingEdges;
-    xid_t xid;    
-    int nIncomingEdges;
-    bool visited;
-} Vertex;
-
-typedef struct Graph
-{
-    Vertex* hashtable[MAX_TRANSACTIONS];
-    Edge* freeEdges;
-    Vertex* freeVertexes;
-} Graph;
+static bool recursiveTraverseGraph(Vertex* root, Vertex* v, int marker);
 
 void initGraph(Graph* graph)
 {
-    memset(graph->hashtable, 0. sizeof(graph->hashtable));
+    memset(graph->hashtable, 0, sizeof(graph->hashtable));
     graph->freeEdges = NULL;
     graph->freeVertexes = NULL;
+    graph->marker = 0;
 }
 
-Edge* newEdge(Graph* graph)
+static inline Edge* newEdge(Graph* graph)
 {
     Edge* edge = graph->freeEdges;
     if (edge == NULL) { 
@@ -44,44 +24,51 @@ Edge* newEdge(Graph* graph)
     return edge;
 }
 
-void freeVertex(Graph* graph, Vertex* vertex)
+static inline void freeVertex(Graph* graph, Vertex* vertex)
 {
-    vertex->node.next = (L2List*)graph->freeVertexes;
+    int h = vertex->xid % MAX_TRANSACTIONS;
+    Vertex** vpp = &graph->hashtable[h];
+    while (*vpp != vertex) { 
+        vpp = &(*vpp)->next;
+    }
+    *vpp = vertex->next;
+    vertex->next = graph->freeVertexes;
     graph->freeVertexes = vertex;
+
 }
 
-void freeEdge(Graph* graph, Edge* edge)
+static inline void freeEdge(Graph* graph, Edge* edge)
 {
     edge->next = graph->freeEdges;
     graph->freeEdges = edge;
 }
 
-Vertex* newVertex(Graph* graph)
+static inline Vertex* newVertex(Graph* graph)
 {
     Vertex* v = graph->freeVertexes;
     if (v == NULL) { 
         v = (Vertex*)malloc(sizeof(Vertex));
     } else { 
-        graph->freeVertexes = (Vertex*)v.node.next;
+        graph->freeVertexes = v->next;
     }
     return v;
 }
 
-Vertex* findVertex(Graph* graph, xid_t xid)
+static inline Vertex* findVertex(Graph* graph, xid_t xid)
 {
-    xid_t h = xid;
+    xid_t h = xid % MAX_TRANSACTIONS;
     Vertex* v;
-    while ((v = graph->hashtable[h % MAX_TRANSACTIONS]) != NULL) { 
+    for (v = graph->hashtable[h]; v != NULL; v = v->next) { 
         if (v->xid == xid) { 
             return v;
         }
-        h += 1;
     }
     v = newVertex(graph);
-    l2_list_init(v->outgoingEdges);
+    l2_list_init(&v->outgoingEdges);
     v->xid = xid;
     v->nIncomingEdges = 0;
-    graph->hashtable[h % MAX_TRANSACTIONS] = v;
+    v->next = graph->hashtable[h];
+    graph->hashtable[h] = v;
     return v;
 }
 
@@ -107,15 +94,38 @@ void addSubgraph(Instance* instance, Graph* graph, xid_t* xids, int n_xids)
         next = e->next;
         l2_list_unlink(&e->node);
         if (--e->dst->nIncomingEdges == 0 && l2_list_is_empty(&e->dst->outgoingEdges)) {
-            freeVertex(e->dst);
+            freeVertex(graph, e->dst);
         }
         if (e->src->nIncomingEdges == 0 && l2_list_is_empty(&e->src->outgoingEdges)) {
-            freeVertex(e->src);
+            freeVertex(graph, e->src);
         }
-        freeEdge(e);
+        freeEdge(graph, e);
     }
 }
 
-bool findLoop(Graph* graph)
+static bool recursiveTraverseGraph(Vertex* root, Vertex* v, int marker)
 {
+    L2List* l;
+    Edge* e;
+    v->visited = marker;
+    for (l = v->outgoingEdges.next; l != &v->outgoingEdges; l = e->node.next) {
+        e = (Edge*)l;
+        if (e->dst == root) { 
+            return true;
+        } else if (e->dst->visited != marker && recursiveTraverseGraph(root, e->dst, marker)) { /* loop */
+            return true;
+        } 
+    }
+    return false;        
+}
+
+bool findLoop(Graph* graph, xid_t root)
+{
+    Vertex* v;
+    for (v = graph->hashtable[root % MAX_TRANSACTIONS]; v != NULL; v = v->next) { 
+        if (v->xid == root) { 
+            return recursiveTraverseGraph(v, v, ++graph->marker);
+        }
+    }
+    return false;        
 }

contrib/pg_dtm/dtmd/include/transaction.h

@@ -500,7 +500,7 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first)
 	return 0;
 }
 
-bool DtmGlobalDetectDeadLock(void* data, int size)
+bool DtmGlobalDetectDeadLock(TransactionId xid, void* data, int size)
 {
     return false;
 }

contrib/pg_dtm/dtmd/src/ddd.c

@@ -59,6 +59,6 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first);
  * Once loop is detected in global resoruce graph, arbiter returns true. Otherwise false is returned.
  * Abiter should replace local part of resource graph if new graph is recevied from this cluster node (not backend).
  */
-bool DtmGlobalDetectDeadLock(void* graph, int size);
+bool DtmGlobalDetectDeadLock(TransactionId xid, void* graph, int size);
 
 #endif

contrib/pg_dtm/libdtm.c

@@ -79,7 +79,7 @@ static TransactionId DtmGetNextXid(void);
 static TransactionId DtmGetNewTransactionId(bool isSubXact);
 static TransactionId DtmGetOldestXmin(Relation rel, bool ignoreVacuum);
 static TransactionId DtmGetGlobalTransactionId(void);
-static bool DtmDetectGlobalDeadLock(void);
+static bool DtmDetectGlobalDeadLock(PGPROC* proc);
 
 static void DtmSerializeLock(PROCLOCK* lock, void* arg);
 
@@ -1023,13 +1023,13 @@ static void DtmSerializeLock(PROCLOCK* proclock, void* arg)
     }
 }
 
-bool DtmDetectGlobalDeadLock(void)
+bool DtmDetectGlobalDeadLock(PGPROC* proc)
 {
     bool hasDeadlock;
     ByteBuffer buf;
     ByteBufferAlloc(&buf);
     EnumerateLocks(DtmSerializeLock, &buf);
-    hasDeadlock = DtmGlobalDetectDeadLock(buf.data, buf.used);
+    hasDeadlock = DtmGlobalDetectDeadLock(proc->lxid, buf.data, buf.used);
     ByteBufferFree(&buf);
     return hasDeadlock;
 }

contrib/pg_dtm/libdtm.h

@@ -23,7 +23,7 @@ TransactionId PgGetGlobalTransactionId(void)
 	return InvalidTransactionId;
 }
 
-bool PgDetectGlobalDeadLock()
+bool PgDetectGlobalDeadLock(PGPROC* proc)
 {
     return false;
 }

contrib/pg_dtm/pg_dtm.c

@@ -11,6 +11,7 @@
 #ifndef XTM_H
 #define XTM_H
 
+#include "storage/proc.h"
 #include "access/clog.h"
 #include "utils/snapmgr.h"
 #include "utils/relcache.h"
@@ -42,7 +43,7 @@ typedef struct
 	bool (*IsInSnapshot)(TransactionId xid, Snapshot snapshot);
 
     /* Detect distributed deadlock */
-    bool (*DetectGlobalDeadLock)(void);
+    bool (*DetectGlobalDeadLock)(PGPROC* proc);
 } TransactionManager;
 
 /* Get pointer to transaction manager: actually returns content of TM variable */
@@ -68,6 +69,6 @@ extern TransactionId PgGetGlobalTransactionId(void);
 
 extern TransactionId PgGetNewTransactionId(bool isSubXact);
 
-extern bool PgDetectGlobalDeadLock(void);
+extern bool PgDetectGlobalDeadLock(PGPROC* proc);
 
 #endif