dmq: add sender connection counters.

Counters allow to notice send connection reset which could possibly drop request
-- and thus we should stop waiting for response. However, they don't save from
potential deadlocks -- if a loop of nodes fills all queues in one direction with
requests (reverse direction must constantly blink for that), nobody will be able
to send replies. So eventually we should have 'connection lost in one direction,
close it in the other' logic. But such deadlocks seem to be hardly possible
in practice, so not today.

gather() is adapted to new API, but nodoby uses it yet. It doesn't rely on
whether node itself and others are enabled/disabled anymore.

This also adds separate DMQ_N_MASK_POS macro, as we can actually have small
number of DMQ_MAX_DESTINATIONS with large mask positions.

Author: arssher

src/commit.c

@@ -334,7 +334,7 @@ MtmTwoPhaseCommit()
 		AllowTempIn2PC = true;
 		CommitTransactionCommand();
 		gtx->state.status = GTXPrepared;
-		gather(participants, (MtmMessage **) messages, &n_messages, false);
+		gather(participants, (MtmMessage **) messages, &n_messages, NULL);
 		dmq_stream_unsubscribe(stream);
 
 		/* check prepare responses */
@@ -381,7 +381,7 @@ MtmTwoPhaseCommit()
 		gtx->state.status = GTXPreCommitted;
 		gtx->state.accepted = (GlobalTxTerm) {1, 0};
 		mtm_log(MtmTxFinish, "TXFINISH: %s precommitted", gid);
-		gather(participants, (MtmMessage **) messages, &n_messages, false);
+		gather(participants, (MtmMessage **) messages, &n_messages, NULL);
 
 		/* check ballots in answers */
 		for (i = 0; i < n_messages; i++)
@@ -442,8 +442,17 @@ MtmTwoPhaseCommit()
 	return true;
 }
 
+/*
+ * For each counterparty in participants, either receives and pushes to
+ * messages a msg from it or waits until recv connection to it dies.
+ * If sendconn_cnt is not NULL, it must contain sender conn counters from
+ * dmq_get_sendconn_cnt. In this case, we stop waiting for counterparty when
+ * notice sender connection reset.
+ * sendconn_cnt must be passed whenever you've sent request via dmq and
+ * expect reply to it, otherwise you risk hanging here infinitely.
+ */
 void
-gather(uint64 participants, MtmMessage **messages, int *msg_count, bool ignore_disabled)
+gather(uint64 participants, MtmMessage **messages, int *msg_count, int *sendconn_cnt)
 {
 	*msg_count = 0;
 	while (participants != 0)
@@ -467,29 +476,12 @@ gather(uint64 participants, MtmMessage **messages, int *msg_count, bool ignore_d
 			mtm_log(MtmTxTrace,
 					"gather: got message from node%d", sender_mask_pos + 1);
 		}
-		else
+		else if (sender_mask_pos != -1)
 		{
-			/*
-			 * If queue is detached then the neignbour node is probably
-			 * disconnected. Let's wait when it became disabled as we can
-			 * became offline by this time.
-			 */
-			int			i;
-			nodemask_t	enabled = MtmGetEnabledNodeMask(ignore_disabled);
-
-			for (i = 0; i < MTM_MAX_NODES; i++)
-			{
-				if (BIT_CHECK(participants, i) && !BIT_CHECK(enabled, i))
-				{
-					BIT_CLEAR(participants, i);
-					mtm_log(MtmTxTrace,
-							"GatherPrecommit: dropping node%d from tx participants",
-							i + 1);
-				}
-			}
+			/* dead recv conn */
+			BIT_CLEAR(participants, sender_mask_pos);
 		}
-
-		if (wait)
+		else /* WOULDBLOCK */
 		{
 			/* XXX cache that */
 			rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT, 100.0,
@@ -498,9 +490,18 @@ gather(uint64 participants, MtmMessage **messages, int *msg_count, bool ignore_d
 			if (rc & WL_POSTMASTER_DEATH)
 				proc_exit(1);
 
+			/* XXX tell the caller about latch reset */
 			if (rc & WL_LATCH_SET)
 				ResetLatch(MyLatch);
+
+			/* waited a bit fruitlessly -- probably sender conn died? */
+			if (rc & WL_TIMEOUT)
+			{
+				participants = dmq_purge_failed_participants(participants,
+															 sendconn_cnt);
+			}
 		}
+
 	}
 }
 
@@ -542,7 +543,7 @@ MtmExplicitPrepare(char *gid)
 
 	mtm_log(MtmTxFinish, "TXFINISH: %s prepared", gid);
 
-	gather(participants, (MtmMessage **) messages, &n_messages, false);
+	gather(participants, (MtmMessage **) messages, &n_messages, NULL);
 	dmq_stream_unsubscribe(stream);
 
 	for (i = 0; i < n_messages; i++)
@@ -593,13 +594,13 @@ MtmExplicitFinishPrepared(bool isTopLevel, char *gid, bool isCommit)
 
 		SetPreparedTransactionState(gid, MULTIMASTER_PRECOMMITTED, false);
 		mtm_log(MtmTxFinish, "TXFINISH: %s precommitted", gid);
-		gather(participants, (MtmMessage **) messages, &n_messages, false);
+		gather(participants, (MtmMessage **) messages, &n_messages, NULL);
 
 		FinishPreparedTransaction(gid, true, false);
 
 		/* XXX: make this conditional */
 		mtm_log(MtmTxFinish, "TXFINISH: %s committed", gid);
-		gather(participants, (MtmMessage **) messages, &n_messages, false);
+		gather(participants, (MtmMessage **) messages, &n_messages, NULL);
 
 		dmq_stream_unsubscribe(gid);
 	}

src/dmq.c

@@ -51,7 +51,8 @@
 #define DMQ_MQ_MAGIC 0x646d71
 
 /*  XXX: move to common */
-#define BIT_CHECK(mask, bit) (((mask) & ((int64)1 << (bit))) != 0)
+#define BIT_CLEAR(mask, bit) ((mask) &= ~((uint64)1 << (bit)))
+#define BIT_CHECK(mask, bit) (((mask) & ((uint64)1 << (bit))) != 0)
 
 /*
  * Shared data structures to hold current connections topology.
@@ -83,6 +84,7 @@ typedef struct
 	PGconn	   *pgconn;
 	DmqConnState state;
 	int			pos;
+	int8		mask_pos;
 } DmqDestination;
 
 typedef struct
@@ -101,6 +103,43 @@ struct DmqSharedState
 	pid_t		sender_pid;
 	dsm_handle	out_dsm;
 	DmqDestination destinations[DMQ_MAX_DESTINATIONS];
+	/*
+	 * Stores counters incremented on each reconnect to destination, indexed
+	 * by receiver mask_pos. This allows to detect conn failures to avoid
+	 * infinite waiting for response when request could have been dropped,
+	 * c.f. dmq_fill_sconn_cnt and dmq_purge_failed_participants.
+	 *
+	 * XXX This mechanism is unreliable and ugly.
+	 * Unreliable, because though it saves from infinite waiting for reply, it
+	 * doesn't save from potential deadlocks. Deadlocks may arise whenever a
+	 * loop of nodes makes request-response sequences because request A->B and
+	 * A's response to B's request go via the same TCP channel; thus, if all
+	 * queues of loop in one direction are filled with requests, nobody will
+	 * be able to answer.
+	 *
+	 * We could control the situation by ensuring 1) all possible requests
+	 * node sends at time could fit in the output buffers 2) node never
+	 * repeats the request until the previous one is delivered or dropped.
+	 * However, to honor the 2), we must terminate send connection whenever
+	 * receive conn failed (and thus we gonna to retry the requests) to flush
+	 * previous possible undelivered requests, which we don't do currently.
+	 *
+	 * Ultimate non-deadlockable solution without such hacks would be to
+	 * divide the job between different channels: A sends its requests to B
+	 * and recevies responses from it via one TCP channel, and B sends its
+	 * requests to A and receives responses via another one. Probably this is
+	 * not worthwhile though as it would make dmq more complicated and
+	 * increase number of shm_mqs.
+	 *
+	 * Besides, the counters are ugly because they require the external code
+	 * to remember sender counters before request and check them while
+	 * waiting for reply; moreover, this checking must be based on timouts
+	 * as nobody would wake the clients on send conn failures.
+	 *
+	 * No locks are used because we don't care much about correct/up-to-date
+	 * reads, though well aligned ints are atomic anyway.
+	 */
+	volatile int sconn_cnt[DMQ_MAX_DESTINATIONS];
 
 	/* receivers stuff */
 	int			n_receivers;
@@ -115,6 +154,9 @@ struct DmqSharedState
 
 }		   *dmq_state;
 
+/* special value for sconn_cnt[] meaning the connection is dead */
+#define DMQSCONN_DEAD 0
+
 static HTAB *dmq_subscriptions;
 
 /* Backend-local i/o queues. */
@@ -332,6 +374,11 @@ dmq_sender_main(Datum main_arg)
 	WaitEventSet *set;
 	DmqDestination conns[DMQ_MAX_DESTINATIONS];
 	int			heartbeat_send_timeout = DatumGetInt32(main_arg);
+	/*
+	 * Seconds dmq_state->sconn_cnt to save the counter value when
+	 * conn is dead.
+	 */
+	int			sconn_cnt[DMQ_MAX_DESTINATIONS];
 
 	double		prev_timer_at = dmq_now();
 
@@ -403,6 +450,8 @@ dmq_sender_main(Datum main_arg)
 					conns[i] = *dest;
 					Assert(conns[i].pgconn == NULL);
 					conns[i].state = Idle;
+					sconn_cnt[dest->mask_pos] = 0;
+					dmq_state->sconn_cnt[dest->mask_pos] = DMQSCONN_DEAD;
 					prev_timer_at = 0;	/* do not wait for timer event */
 				}
 				/* close connection to deleted destination */
@@ -443,6 +492,7 @@ dmq_sender_main(Datum main_arg)
 					if (ret < 0)
 					{
 						conns[conn_id].state = Idle;
+						dmq_state->sconn_cnt[conns[conn_id].mask_pos] = DMQSCONN_DEAD;
 
 						mtm_log(DmqStateFinal,
 								"[DMQ] failed to send message to %s: %s",
@@ -561,6 +611,7 @@ dmq_sender_main(Datum main_arg)
 					if (ret < 0)
 					{
 						conns[conn_id].state = Idle;
+						dmq_state->sconn_cnt[conns[conn_id].mask_pos] = DMQSCONN_DEAD;
 
 						mtm_log(DmqStateFinal,
 								"[DMQ] failed to send heartbeat to %s: %s",
@@ -684,9 +735,13 @@ dmq_sender_main(Datum main_arg)
 					}
 					if (!PQisBusy(conns[conn_id].pgconn))
 					{
+						int8 mask_pos = conns[conn_id].mask_pos;
+
 						conns[conn_id].state = Active;
 						DeleteWaitEvent(set, event.pos);
 						PQsetnonblocking(conns[conn_id].pgconn, 1);
+						sconn_cnt[mask_pos]++;
+						dmq_state->sconn_cnt[mask_pos] = sconn_cnt[mask_pos];
 
 						mtm_log(DmqStateFinal,
 								"[DMQ] Connected to %s",
@@ -702,6 +757,7 @@ dmq_sender_main(Datum main_arg)
 					if (!PQconsumeInput(conns[conn_id].pgconn))
 					{
 						conns[conn_id].state = Idle;
+						dmq_state->sconn_cnt[conns[conn_id].mask_pos] = DMQSCONN_DEAD;
 
 						mtm_log(DmqStateFinal,
 								"[DMQ] connection error with %s: %s",
@@ -1444,6 +1500,9 @@ dmq_detach_receiver(char *sender_name)
 	dmq_local.inhandles[handle_id].name[0] = '\0';
 }
 
+/*
+ * Subscribes caller to msgs from stream_name.
+ */
 void
 dmq_stream_subscribe(char *stream_name)
 {
@@ -1503,6 +1562,24 @@ dmq_stream_unsubscribe(char *stream_name)
 	Assert(found);
 }
 
+/*
+ * Fills (preallocated) sconn_cnt with current values of sender
+ * connection counters to guys in participants mask (as registered in
+ * dmq_destination_add), which allows to reveal connection failures, possibly
+ * resulted in losing request -- and thus stop hopeless waiting for response.
+ */
+void
+dmq_get_sendconn_cnt(uint64 participants, int *sconn_cnt)
+{
+	int i;
+
+	for (i = 0; i < DMQ_N_MASK_POS; i++)
+	{
+		if (BIT_CHECK(participants, i))
+			sconn_cnt[i] = dmq_state->sconn_cnt[i];
+	}
+}
+
 bool
 dmq_pop(int8 *sender_mask_pos, StringInfo msg, uint64 mask)
 {
@@ -1639,9 +1716,36 @@ dmq_pop_nb(int8 *sender_mask_pos, StringInfo msg, uint64 mask, bool *wait)
 	return false;
 }
 
+/*
+ * Accepts bitmask of participants and sconn_cnt counters with send
+ * connections counters as passed to (and the latter filled by)
+ * dmq_fill_sconn_cnt, returns this mask after unsetting bits for those
+ * counterparties with whom we've lost the send connection since.
+ */
+uint64
+dmq_purge_failed_participants(uint64 participants, int *sconn_cnt)
+{
+	int i;
+	uint64 res = participants;
+
+	for (i = 0; i < DMQ_N_MASK_POS; i++)
+	{
+		if (BIT_CHECK(participants, i) &&
+			(sconn_cnt[i] == DMQSCONN_DEAD ||
+			 sconn_cnt[i] != dmq_state->sconn_cnt[i]))
+			BIT_CLEAR(res, i);
+	}
+	return res;
+}
+
+/*
+ * recv_mask_pos is short (< DMQ_N_MASK_POS) variant of
+ * receiver_name, used to track connection failures -- it must match mask_pos
+ * in dmq_attach_receiver to work!
+ */
 DmqDestinationId
 dmq_destination_add(char *connstr, char *sender_name, char *receiver_name,
-					int recv_timeout)
+					int8 recv_mask_pos, int recv_timeout)
 {
 	DmqDestinationId dest_id;
 	pid_t		sender_pid;
@@ -1658,6 +1762,7 @@ dmq_destination_add(char *connstr, char *sender_name, char *receiver_name,
 			strncpy(dest->connstr, connstr, DMQ_CONNSTR_MAX_LEN);
 			dest->recv_timeout = recv_timeout;
 			dest->active = true;
+			dest->mask_pos = recv_mask_pos;
 			break;
 		}
 	}

src/include/commit.h

@@ -28,6 +28,6 @@ extern void MtmXactCallback(XactEvent event, void *arg);
 extern bool MtmExplicitPrepare(char *gid);
 extern void MtmExplicitFinishPrepared(bool isTopLevel, char *gid, bool isCommit);
 
-extern void gather(uint64 participants, MtmMessage **messages, int *msg_count, bool ignore_mtm_disabled);
+extern void gather(uint64 participants, MtmMessage **messages, int *msg_count, int *sendconn_cnt);
 
 #endif

src/include/dmq.h

@@ -10,8 +10,10 @@ typedef int8 DmqDestinationId;
 
 extern void dmq_init(int send_timeout);
 
+#define DMQ_N_MASK_POS 16 /* ought to be >= MTM_MAX_NODES */
 extern DmqDestinationId dmq_destination_add(char *connstr, char *sender_name,
-					char *receiver_name, int ping_period);
+											char *receiver_name, int8 recv_mask_pos,
+											int ping_period);
 extern void dmq_destination_drop(char *receiver_name);
 
 extern void dmq_attach_receiver(char *sender_name, int8 mask_pos);
@@ -22,8 +24,10 @@ extern void dmq_terminate_receiver(char *name);
 extern void dmq_stream_subscribe(char *stream_name);
 extern void dmq_stream_unsubscribe(char *stream_name);
 
+extern void dmq_get_sendconn_cnt(uint64 participants, int *sconn_cnt);
 extern bool dmq_pop(int8 *sender_mask_pos, StringInfo msg, uint64 mask);
 extern bool dmq_pop_nb(int8 *sender_mask_pos, StringInfo msg, uint64 mask, bool *wait);
+extern uint64 dmq_purge_failed_participants(uint64 participants, int *sconn_cnt);
 
 extern void dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg);
 extern void dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *buffer, size_t len);

src/resolver.c

@@ -204,7 +204,7 @@ scatter_status_requests(MtmConfig *mtm_cfg)
 		 * ars: we might collect here responses from previous resolving
 		 * sessions, which would fill acks with non MtmLastTermResponse messages.
 		 */
-		gather(connected, (MtmMessage **) acks, &n_acks, true);
+		gather(connected, (MtmMessage **) acks, &n_acks, NULL);
 
 		for (i = 0; i < n_acks; i++)
 		{

src/state.c

@@ -1471,7 +1471,7 @@ start_node_workers(int node_id, MtmConfig *new_cfg, Datum arg)
 
 	/* Add dmq destination */
 	dest = dmq_destination_add(dmq_connstr, dmq_my_name, dmq_node_name,
-							   MtmHeartbeatRecvTimeout);
+							   node_id - 1, MtmHeartbeatRecvTimeout);
 
 	LWLockAcquire(Mtm->lock, LW_EXCLUSIVE);
 	Mtm->peers[node_id - 1].dmq_dest_id = dest;