Integrate deadlock detection in DTMD

Author: knizhnik

contrib/pg_dtm/dtmd/include/proto.h

@@ -7,9 +7,11 @@
 #define CMD_AGAINST  'n'
 #define CMD_SNAPSHOT 'h'
 #define CMD_STATUS   's'
+#define CMD_DEADLOCK 'd'
 
 #define RES_FAILED 0xDEADBEEF
 #define RES_OK 0xC0FFEE
+#define RES_DEADLOCK 0xDEADDEED
 #define RES_TRANSACTION_COMMITTED 1
 #define RES_TRANSACTION_ABORTED 2
 #define RES_TRANSACTION_INPROGRESS 3

contrib/pg_dtm/dtmd/src/main.c

@@ -13,6 +13,7 @@
 #include "util.h"
 #include "transaction.h"
 #include "proto.h"
+#include "ddd.h"
 
 #define DEFAULT_DATADIR "/tmp/clog"
 #define DEFAULT_LISTENHOST "0.0.0.0"
@@ -32,6 +33,7 @@ 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;
@@ -158,6 +160,7 @@ static void debug_cmd(client_t client, int argc, xid_t *argv) {
 		case CMD_AGAINST : cmdname =  "AGAINST"; break;
 		case CMD_SNAPSHOT: cmdname = "SNAPSHOT"; break;
 		case CMD_STATUS  : cmdname =   "STATUS"; break;
+		case CMD_DEADLOCK: cmdname = "DEADLOCK"; break;
 		default          : cmdname =  "unknown";
 	}
 	debug("[%d] %s", CLIENT_ID(client), cmdname);
@@ -530,6 +533,25 @@ static void onnoise(client_t client, int argc, xid_t *argv) {
 	client_message_shortcut(client, RES_FAILED);
 }
 
+static Graph graph;
+
+static void ondeadlock(client_t client, int argc, xid_t *argv) {
+	if (argc < 3) {
+		shout(
+			"[%d] DEADLOCK: wrong number of arguments %d, expected > 3\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);
+    bool hasDeadLock = findLoop(&graph, root);
+    client_message_shortcut(client, hasDeadLock ? RES_DEADLOCK : RES_OK);
+}
+    
+
 static void oncmd(client_t client, int argc, xid_t *argv) {
 	debug_cmd(client, argc, argv);
 
@@ -553,6 +575,9 @@ static void oncmd(client_t client, int argc, xid_t *argv) {
 		case CMD_STATUS:
 			onstatus(client, argc, argv);
 			break;
+		case CMD_DEADLOCK:
+			ondeadlock(client, argc, argv);
+			break;
 		default:
 			onnoise(client, argc, argv);
 	}

contrib/pg_dtm/dtmd/src/util.c

@@ -31,7 +31,7 @@ char *join_path(const char *dir, const char *file) {
 		}
 	} else {
 		// 'file' is empty
-		return strndup(dir, dirlen);
+		return strdup(dir);
 	}
 
 	size_t pathlen = dirlen + 1 + filelen;

contrib/pg_dtm/libdtm.c

@@ -502,5 +502,39 @@ int DtmGlobalReserve(TransactionId xid, int nXids, TransactionId *first)
 
 bool DtmGlobalDetectDeadLock(TransactionId xid, void* data, int size)
 {
-    return false;
+	int msg_size = size + sizeof(xid)*2;
+    int data_size = sizeof(ShubMessageHdr) + msg_size;
+    char* buf = (char*)malloc(data_size);
+	ShubMessageHdr* msg = (ShubMessageHdr*)buf;
+	xid_t* body = (xid_t*)(msg+1);
+    int sent;
+	int reslen;
+	xid_t results[RESULTS_SIZE];
+	DTMConn dtm = GetConnection();
+
+	msg->chan = 0;
+	msg->code = MSG_FIRST_USER_CODE;
+	msg->size = msg_size;
+
+	*body++ = CMD_DEADLOCK;
+    *body++ = xid;
+    memcpy(body, data, size);
+
+	sent = 0;
+	while (sent < data_size)
+	{
+		int new_bytes = write(dtm->sock, buf + sent, data_size - sent);
+		if (new_bytes == -1)
+		{
+			elog(ERROR, "Failed to send a command to arbiter");
+			return false;
+		}
+		sent += new_bytes;
+	}
+	reslen = dtm_recv_results(dtm, RESULTS_SIZE, results);
+	if (reslen != 1 || (results[0] != RES_OK && results[0] != RES_DEADLOCK)) { 
+        fprintf(stderr, "DtmGlobalDetectDeadLock: failed to check deadlocks for transaction %u\n", xid);
+        return false;
+    }
+	return results[0] == RES_DEADLOCK;
 }