Add node hash table

Author: knizhnik

contrib/pg_dtm/dtmd/include/ddd.h

@@ -3,9 +3,11 @@
 #include <stdbool.h>
 #include "transaction.h"
 
-typedef struct Instance {
+typedef struct Node {
+    struct Node* collision;
     struct Edge* edges; /* local subgraph */    
-} Instance;
+    nodeid_t node_id;
+} Node;
 
 typedef struct Edge {
     L2List node; /* node of list of outgoing eedges */
@@ -33,9 +35,13 @@ typedef struct Graph
     int min_deadlock_duration;
 } Graph;
 
+typedef struct Cluster 
+{
+    Node* hashtable[MAX_STREAMS];
+} Cluster;
 
 extern void initGraph(Graph* graph);
-extern void addSubgraph(Instance* instance, Graph* graph, xid_t* xids, int n_xids);
+extern void addSubgraph(Graph* graph, nodeid_t node_id, xid_t* xids, int n_xids);
 extern bool detectDeadLock(Graph* graph, xid_t root);
 
 #endif

contrib/pg_dtm/dtmd/include/int.h

@@ -2,5 +2,6 @@
 #define INT_H
 
 typedef unsigned xid_t;
+typedef unsigned long long nodeid_t;
 
 #endif

contrib/pg_dtm/dtmd/include/server.h

@@ -98,4 +98,6 @@ bool client_message_finish(client_t client);
  */
 bool client_message_shortcut(client_t client, xid_t arg);
 
+unsigned client_get_ip_addr(client_t client);
+
 #endif

contrib/pg_dtm/dtmd/src/ddd.c

@@ -5,6 +5,8 @@
 
 static bool recursiveTraverseGraph(Vertex* root, Vertex* v, int marker);
 
+static Cluster cluster;
+
 void initGraph(Graph* graph)
 {
     memset(graph->hashtable, 0, sizeof(graph->hashtable));
@@ -74,10 +76,28 @@ static inline Vertex* findVertex(Graph* graph, xid_t xid)
     return v;
 }
 
-void addSubgraph(Instance* instance, Graph* graph, xid_t* xids, int n_xids)
+static inline Node* findNode(Cluster* cluster, nodeid_t node_id)
+{
+    size_t h = node_id % MAX_STREAMS;
+    Node* node;
+    for (node = cluster->hashtable[h]; node != NULL; node = node->collision) { 
+        if (node->node_id == node_id) { 
+            return node;
+        }
+    }
+    node = (Node*)malloc(sizeof(Node));
+    node->node_id = node_id;
+    node->edges = NULL;
+    node->collision = cluster->hashtable[h];
+    cluster->hashtable[h] = node;
+    return node;
+}
+
+void addSubgraph(Graph* graph, nodeid_t node_id, xid_t* xids, int n_xids)
 {
     xid_t *last = xids + n_xids;
     Edge *e, *next, *edges = NULL;
+    Node* node = findNode(&cluster, node_id);
     while (xids != last) { 
         Vertex* src = findVertex(graph, *xids++);
         xid_t xid;
@@ -92,7 +112,7 @@ void addSubgraph(Instance* instance, Graph* graph, xid_t* xids, int n_xids)
             l2_list_link(&src->outgoingEdges, &e->node);
         }
     }
-    for (e = instance->edges; e != NULL; e = next) { 
+    for (e = node->edges; e != NULL; e = next) { 
         next = e->next;
         l2_list_unlink(&e->node);
         if (--e->dst->nIncomingEdges == 0 && l2_list_is_empty(&e->dst->outgoingEdges)) {
@@ -103,7 +123,7 @@ void addSubgraph(Instance* instance, Graph* graph, xid_t* xids, int n_xids)
         }
         freeEdge(graph, e);
     }
-    instance->edges = edges;
+    node->edges = edges;
 }
 
 static bool recursiveTraverseGraph(Vertex* root, Vertex* v, int marker)

contrib/pg_dtm/dtmd/src/main.c

@@ -33,7 +33,6 @@ typedef struct client_userdata_t {
 	int id;
 	int snapshots_sent;
 	xid_t xid;
-    Instance instance; /* It has to be moved somewhere else, because this is per-backend structure */
 } client_userdata_t;
 
 clog_t clg;
@@ -536,17 +535,22 @@ static void onnoise(client_t client, int argc, xid_t *argv) {
 static Graph graph;
 
 static void ondeadlock(client_t client, int argc, xid_t *argv) {
-	if (argc < 3) {
+    int port;
+    xid_t root;
+    nodeid_t node_id;
+
+	if (argc < 4) {
 		shout(
-			"[%d] DEADLOCK: wrong number of arguments %d, expected > 3\n",
+			"[%d] DEADLOCK: wrong number of arguments %d, expected > 4\n",
 			CLIENT_ID(client), argc
 		);        
 		client_message_shortcut(client, RES_FAILED);
 		return;
 	}
-    xid_t root = argv[1];
-    Instance* instance = &CLIENT_USERDATA(client)->instance;
-    addSubgraph(instance, &graph, argv+2, argc-2);
+    port = argv[1];
+    root = argv[2];
+    node_id = ((nodeid_t)port << 32) | client_get_ip_addr(client);
+    addSubgraph(&graph, node_id, argv+3, argc-3);
     bool hasDeadLock = detectDeadLock(&graph, root);
     client_message_shortcut(client, hasDeadLock ? RES_DEADLOCK : RES_OK);
 }

contrib/pg_dtm/dtmd/src/server.c

@@ -14,6 +14,7 @@
 #include <sys/socket.h>
 #include <arpa/inet.h>
 #include <netinet/tcp.h>
+#include <netinet/in.h>
 
 #include "server.h"
 #include "limits.h"
@@ -490,6 +491,15 @@ void *client_get_userdata(client_t client) {
 	return client->userdata;
 }
 
+unsigned client_get_ip_addr(client_t client)
+{
+    struct sockaddr_in inet_addr;
+    socklen_t inet_addr_len = sizeof(inet_addr);
+    inet_addr.sin_addr.s_addr = 0;
+    getpeername(client->stream->fd, (struct sockaddr *)&inet_addr, &inet_addr_len);
+    return inet_addr.sin_addr.s_addr;
+}
+   
 #if 0
 // usage example
 

contrib/pg_dtm/libdtm.c

@@ -481,13 +481,13 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first)
 	return 0;
 }
 
-bool DtmGlobalDetectDeadLock(TransactionId xid, void* data, int size)
+bool DtmGlobalDetectDeadLock(int port, TransactionId xid, void* data, int size)
 {
-	int msg_size = size + sizeof(xid)*2;
+	int msg_size = size + sizeof(xid)*3;
 	int data_size = sizeof(ShubMessageHdr) + msg_size;
 	char* buf = (char*)malloc(data_size);
 	ShubMessageHdr* msg = (ShubMessageHdr*)buf;
-	xid_t* body = (xid_t*)(msg+1);
+	xid_t* body = (xid_t*)(msg+2);
 	int sent;
 	int reslen;
 	xid_t results[RESULTS_SIZE];
@@ -498,6 +498,7 @@ bool DtmGlobalDetectDeadLock(TransactionId xid, void* data, int size)
 	msg->size = msg_size;
 
 	*body++ = CMD_DEADLOCK;
+	*body++ = port;
 	*body++ = xid;
 	memcpy(body, data, size);
 

contrib/pg_dtm/libdtm.h

@@ -62,6 +62,6 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first);
  * resource graph if a new local graph is received from this cluster node (not
  * backend).
  */
-bool DtmGlobalDetectDeadLock(TransactionId xid, void* graph, int size);
+bool DtmGlobalDetectDeadLock(int port, TransactionId xid, void* graph, int size);
 
 #endif

contrib/pg_dtm/pg_dtm.c

@@ -1029,7 +1029,7 @@ bool DtmDetectGlobalDeadLock(PGPROC* proc)
     ByteBuffer buf;
     ByteBufferAlloc(&buf);
     EnumerateLocks(DtmSerializeLock, &buf);
-    hasDeadlock = DtmGlobalDetectDeadLock(proc->lxid, buf.data, buf.used);
+    hasDeadlock = DtmGlobalDetectDeadLock(PostPortNumber, proc->lxid, buf.data, buf.used);
     ByteBufferFree(&buf);
     return hasDeadlock;
 }