Fix a silly deadlock in DTMD/Raft. Add 'hello' message to the protocol.

Author: kvap

contrib/pg_dtm/README

@@ -43,6 +43,13 @@ The format of all commands:
 
 The commands:
 
+'h': hello()
+	The first message.
+
+	The arbiter replies with:
+		[RES_OK] if ready
+		[RES_FAILED] (or disconnection) if not ready
+
 'r': reserve(minxid, minsize)
 	Claims a sequence ≥ minsize of xids ≥ minxid for local usage. This will
 	prevent the arbiter from using those values for global transactions.
@@ -88,7 +95,7 @@ The commands:
 
 	The reply and 'wait' logic is the same as for the 'status' command.
 
-'h': snapshot(xid)
+'t': snapshot(xid)
 	Tells the arbiter to generate a snapshot for the global transaction
 	identified by the given 'xid'. The arbiter will create a snapshot for
 	every participant, so when each of them asks for the snapshot it will

contrib/pg_dtm/dtmd/include/dtmdlimits.h

@@ -11,7 +11,7 @@
 #define HEARTBEAT_TIMEOUT_MS 20
 #define ELECTION_TIMEOUT_MS_MIN 150
 #define ELECTION_TIMEOUT_MS_MAX 300
-#define RAFT_LOGLEN 10240
-#define RAFT_KEEP_APPLIED 4096 // how many applied entries to keep during compaction
+#define RAFT_LOGLEN 1024
+#define RAFT_KEEP_APPLIED 512 // how many applied entries to keep during compaction
 
 #endif

contrib/pg_dtm/dtmd/include/proto.h

@@ -1,11 +1,12 @@
 #ifndef PROTO_H
 #define PROTO_H
 
+#define CMD_HELLO    'h'
 #define CMD_RESERVE  'r'
 #define CMD_BEGIN    'b'
 #define CMD_FOR      'y'
 #define CMD_AGAINST  'n'
-#define CMD_SNAPSHOT 'h'
+#define CMD_SNAPSHOT 't'
 #define CMD_STATUS   's'
 
 #define RES_FAILED 0xDEADBEEF

contrib/pg_dtm/dtmd/src/main.c

@@ -82,6 +82,7 @@ static void notify_listeners(Transaction *t, int status) {
 		// notify 'status' listeners about the transaction status
 		case BLANK:
 			while ((listener = transaction_pop_listener(t, 's'))) {
+				debug("[%d] notifying the client about xid=%u (unknown)\n", CLIENT_ID(listener), t->xid);
 				client_message_shortcut(
 					(client_t)listener,
 					RES_TRANSACTION_UNKNOWN
@@ -90,7 +91,7 @@ static void notify_listeners(Transaction *t, int status) {
 			break;
 		case NEGATIVE:
 			while ((listener = transaction_pop_listener(t, 's'))) {
-				// notify 'status' listeners about the transaction status
+				debug("[%d] notifying the client about xid=%u (aborted)\n", CLIENT_ID(listener), t->xid);
 				client_message_shortcut(
 					(client_t)listener,
 					RES_TRANSACTION_ABORTED
@@ -99,7 +100,7 @@ static void notify_listeners(Transaction *t, int status) {
 			break;
 		case POSITIVE:
 			while ((listener = transaction_pop_listener(t, 's'))) {
-				// notify 'status' listeners about the transaction status
+				debug("[%d] notifying the client about xid=%u (committed)\n", CLIENT_ID(listener), t->xid);
 				client_message_shortcut(
 					(client_t)listener,
 					RES_TRANSACTION_COMMITTED
@@ -108,7 +109,7 @@ static void notify_listeners(Transaction *t, int status) {
 			break;
 		case DOUBT:
 			while ((listener = transaction_pop_listener(t, 's'))) {
-				// notify 'status' listeners about the transaction status
+				debug("[%d] notifying the client about xid=%u (inprogress)\n", CLIENT_ID(listener), t->xid);
 				client_message_shortcut(
 					(client_t)listener,
 					RES_TRANSACTION_INPROGRESS
@@ -122,19 +123,22 @@ static void apply_clog_update(int action, int argument) {
 	int status = action;
 	xid_t xid = argument;
 	assert((status == NEGATIVE) || (status == POSITIVE));
+	debug("APPLYING: xid=%u, status=%d\n", xid, status);
 
 	if (!clog_write(clg, xid, status)) {
-		shout("APPLY: failed to write to clog, xid=%d\n", xid);
+		shout("APPLY: failed to write to clog, xid=%u\n", xid);
 	}
 
-	Transaction *t = find_transaction(xid);
-	if (t == NULL) {
-		debug("APPLY: xid %u is not active\n", xid);
-		return;
-	}
+	if (!use_raft || (raft.role == ROLE_LEADER)) {
+		Transaction *t = find_transaction(xid);
+		if (t == NULL) {
+			debug("APPLY: xid=%u is not active\n", xid);
+			return;
+		}
 
-	notify_listeners(t, status);
-	free_transaction(t);
+		notify_listeners(t, status);
+		free_transaction(t);
+	}
 }
 
 static int next_client_id = 0;
@@ -153,14 +157,16 @@ static void ondisconnect(client_t client) {
 		// need to abort the transaction this client is participating in
 		for (t = (Transaction*)active_transactions.next; t != (Transaction*)&active_transactions; t = (Transaction*)t->elem.next) {
 			if (t->xid == CLIENT_XID(client)) {
-				raft_emit(&raft, NEGATIVE, t->xid);
+				if (use_raft && (raft.role == ROLE_LEADER)) {
+					raft_emit(&raft, NEGATIVE, t->xid);
+				}
 				break;
 			}
 		}
 
 		if (t == (Transaction*)&active_transactions) {
 			shout(
-				"[%d] DISCONNECT: transaction %u not found O_o\n",
+				"[%d] DISCONNECT: transaction xid=%u not found O_o\n",
 				CLIENT_ID(client), CLIENT_XID(client)
 			);
 		}
@@ -175,6 +181,7 @@ static void debug_cmd(client_t client, int argc, xid_t *argv) {
 	char *cmdname;
 	assert(argc > 0);
 	switch (argv[0]) {
+		case CMD_HELLO   : cmdname =    "HELLO"; break;
 		case CMD_RESERVE : cmdname =  "RESERVE"; break;
 		case CMD_BEGIN   : cmdname =    "BEGIN"; break;
 		case CMD_FOR     : cmdname =      "FOR"; break;
@@ -203,6 +210,9 @@ static void debug_cmd(client_t client, int argc, xid_t *argv) {
 		} \
 	} while (0)
 
+#define CHECKLEADER(CLIENT) \
+	CHECK(raft.role == ROLE_LEADER, CLIENT, "not a leader")
+
 static xid_t max_of_xids(xid_t a, xid_t b) {
 	return a > b ? a : b;
 }
@@ -232,6 +242,17 @@ static void gen_snapshot(Snapshot *s) {
 	}
 }
 
+static void onhello(client_t client, int argc, xid_t *argv) {
+	CHECK(argc == 1, client, "HELLO: wrong number of arguments");
+
+	debug("[%d] HELLO\n", CLIENT_ID(client));
+	if (raft.role == ROLE_LEADER) {
+		client_message_shortcut(client, RES_OK);
+	} else {
+		client_message_shortcut(client, RES_FAILED);
+	}
+}
+
 static void onreserve(client_t client, int argc, xid_t *argv) {
 	CHECK(argc == 3, client, "RESERVE: wrong number of arguments");
 
@@ -344,10 +365,12 @@ static void onbegin(client_t client, int argc, xid_t *argv) {
 static bool queue_for_transaction_finish(client_t client, xid_t xid, char cmd) {
 	assert((cmd >= 'a') && (cmd <= 'z'));
 
+	debug("[%d] QUEUE for xid=%u status\n", CLIENT_ID(client), xid);
+
 	Transaction *t = find_transaction(xid);
 	if (t == NULL) {
 		shout(
-			"[%d] QUEUE: xid %u not found\n",
+			"[%d] QUEUE: xid=%u not found\n",
 			CLIENT_ID(client), xid
 		);
 		client_message_shortcut(client, RES_FAILED);
@@ -378,7 +401,7 @@ static void onvote(client_t client, int argc, xid_t *argv, int vote) {
 	Transaction *t = find_transaction(xid);
 	if (t == NULL) {
 		shout(
-			"[%d] VOTE: xid %u not found\n",
+			"[%d] VOTE: xid=%u not found\n",
 			CLIENT_ID(client), xid
 		);
 		client_message_shortcut(client, RES_FAILED);
@@ -445,7 +468,7 @@ static void onsnapshot(client_t client, int argc, xid_t *argv) {
 	Transaction *t = find_transaction(xid);
 	if (t == NULL) {
 		shout(
-			"[%d] SNAPSHOT: xid %u not found\n",
+			"[%d] SNAPSHOT: xid=%u not found\n",
 			CLIENT_ID(client), xid
 		);
 		client_message_shortcut(client, RES_FAILED);
@@ -543,22 +566,31 @@ static void oncmd(client_t client, int argc, xid_t *argv) {
 
 	assert(argc > 0);
 	switch (argv[0]) {
+		case CMD_HELLO:
+			onhello(client, argc, argv);
+			break;
 		case CMD_RESERVE:
+			CHECKLEADER(client);
 			onreserve(client, argc, argv);
 			break;
 		case CMD_BEGIN:
+			CHECKLEADER(client);
 			onbegin(client, argc, argv);
 			break;
 		case CMD_FOR:
+			CHECKLEADER(client);
 			onvote(client, argc, argv, POSITIVE);
 			break;
 		case CMD_AGAINST:
+			CHECKLEADER(client);
 			onvote(client, argc, argv, NEGATIVE);
 			break;
 		case CMD_SNAPSHOT:
+			CHECKLEADER(client);
 			onsnapshot(client, argc, argv);
 			break;
 		case CMD_STATUS:
+			CHECKLEADER(client);
 			onstatus(client, argc, argv);
 			break;
 		default:

contrib/pg_dtm/dtmd/src/raft.c

@@ -143,7 +143,7 @@ int raft_create_udp_socket(raft_t *r) {
 		return -1;
 	}
 	me->addr.sin_port = htons(me->port);
-	debug("binding %s:%d\n", me->host, me->port);
+	debug("binding udp %s:%d\n", me->host, me->port);
 	if (bind(r->sock, (struct sockaddr*)&me->addr, sizeof(me->addr)) == -1) {
 		shout("cannot bind the socket: %s\n", strerror(errno));
 		return -1;
@@ -321,7 +321,6 @@ static int raft_log_compact(raft_log_t *l, int keep_applied) {
 			snap.minarg = min(snap.minarg, e->argument);
 			snap.maxarg = max(snap.maxarg, e->argument);
 		}
-		e->snapshot = false; // FIXME: should not need this, find the code where it is not set on new entry insertion
 		compacted++;
 	}
 	if (compacted) {
@@ -339,7 +338,7 @@ bool raft_emit(raft_t *r, int action, int argument) {
 
 	if (r->log.size == RAFT_LOGLEN) {
 		int compacted = raft_log_compact(&r->log, RAFT_KEEP_APPLIED);
-		if (compacted) {
+		if (compacted > 1) {
 			shout("compacted %d entries\n", compacted);
 		} else {
 			shout(
@@ -348,10 +347,9 @@ bool raft_emit(raft_t *r, int action, int argument) {
 			);
 			return false;
 		}
-		return false;
 	}
 
-	raft_entry_t *e = &RAFT_LOG(r, r->log.size);
+	raft_entry_t *e = &RAFT_LOG(r, r->log.first + r->log.size);
 	e->snapshot = false;
 	e->term = r->term;
 	e->action = action;

contrib/pg_dtm/dtmd/src/server.c

@@ -85,7 +85,7 @@ static int create_listening_socket(const char *host, int port) {
 		return -1;
 	}
 	addr.sin_port = htons(port);
-	debug("binding %s:%d\n", host, port);
+	debug("binding tcp %s:%d\n", host, port);
 	if (bind(s, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
 		shout("cannot bind the listening socket: %s\n", strerror(errno));
 		return -1;