more node checks during gather

Author: kelvich

src/commit.c

@@ -22,6 +22,8 @@
 #include "tcop/tcopprot.h"
 #include "postmaster/autovacuum.h"
 #include "libpq/pqformat.h"
+#include "pgstat.h"
+#include "storage/ipc.h"
 
 #include "multimaster.h"
 #include "logger.h"
@@ -32,7 +34,7 @@
 
 typedef struct
 {
-	StringInfo	message;
+	StringInfoData	message;
 	int			node_id;
 } mtm_msg;
 
@@ -267,7 +269,7 @@ MtmTwoPhaseCommit()
 
 	for (i = 0; i < n_messages; i++)
 	{
-		MtmMessageCode status = pq_getmsgbyte(messages[i].message);
+		MtmMessageCode status = pq_getmsgbyte(&messages[i].message);
 
 		Assert(status == MSG_PREPARED || status == MSG_ABORTED);
 		if (status == MSG_ABORTED)
@@ -307,11 +309,13 @@ gather(uint64 participants, mtm_msg *messages, int *msg_count)
 	*msg_count = 0;
 	while (participants != 0)
 	{
-		bool ret;
-		DmqSenderId sender_id;
-		StringInfo msg = makeStringInfo();
+		bool		ret;
+		DmqSenderId	sender_id;
+		StringInfoData	msg;
+		int			rc;
+		bool		wait;
 
-		ret = dmq_pop(&sender_id, msg, participants);
+		ret = dmq_pop_nb(&sender_id, &msg, participants, &wait);
 		if (ret)
 		{
 			messages[*msg_count].message = msg;
@@ -323,7 +327,7 @@ gather(uint64 participants, mtm_msg *messages, int *msg_count)
 					"gather: got message from node%d",
 					sender_to_node[sender_id]);
 		}
-		else
+		else if (sender_id >= 0)
 		{
 			/*
 			 * If queue is detached then the neignbour node is probably
@@ -338,6 +342,19 @@ gather(uint64 participants, mtm_msg *messages, int *msg_count)
 					sender_to_node[sender_id]);
 			}
 		}
+
+		if (wait)
+		{
+			// XXX cache that
+			rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_TIMEOUT, 100.0,
+						PG_WAIT_EXTENSION);
+
+			if (rc & WL_POSTMASTER_DEATH)
+				proc_exit(1);
+
+			if (rc & WL_LATCH_SET)
+				ResetLatch(MyLatch);
+		}
 	}
 }
 
@@ -373,7 +390,7 @@ MtmExplicitPrepare(char *gid)
 
 	for (i = 0; i < n_messages; i++)
 	{
-		MtmMessageCode status = pq_getmsgbyte(messages[i].message);
+		MtmMessageCode status = pq_getmsgbyte(&messages[i].message);
 
 		Assert(status == MSG_PREPARED || status == MSG_ABORTED);
 		if (status == MSG_ABORTED)

src/dmq.c

@@ -1529,11 +1529,14 @@ dmq_pop(DmqSenderId *sender_id, StringInfo msg, uint64 mask)
 }
 
 bool
-dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask)
+dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask, bool *wait)
 {
 	shm_mq_result res;
 	int i;
 
+	*wait = true;
+	*sender_id = -1;
+
 	for (i = 0; i < dmq_local.n_inhandles; i++)
 	{
 		Size	len;
@@ -1553,7 +1556,9 @@ dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask)
 			msg->len = len;
 			msg->maxlen = -1;
 			msg->cursor = 0;
+
 			*sender_id = i;
+			*wait = false;
 
 			mtm_log(DmqTraceIncoming,
 					"[DMQ] dmq_pop_nb: got message %s (len=%zu) from %s",
@@ -1562,8 +1567,11 @@ dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask)
 		}
 		else if (res == SHM_MQ_DETACHED)
 		{
-			if (!dmq_reattach_shm_mq(i))
-				mtm_log(WARNING, "[DMQ] dmq_pop_nb: failed to reattach");
+			*sender_id = i;
+
+			if (dmq_reattach_shm_mq(i))
+				*wait = false;
+
 			return false;
 		}
 	}

src/include/dmq.h

@@ -24,7 +24,7 @@ extern void dmq_stream_subscribe(char *stream_name);
 extern void dmq_stream_unsubscribe(char *stream_name);
 
 extern bool dmq_pop(DmqSenderId *sender_id, StringInfo msg, uint64 mask);
-extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask);
+extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask, bool *wait);
 
 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

@@ -467,8 +467,9 @@ handle_responses(void)
 {
 	DmqSenderId sender_id;
 	StringInfoData msg;
+	bool		wait;
 
-	while(dmq_pop_nb(&sender_id, &msg, MtmGetConnectedNodeMask()))
+	while(dmq_pop_nb(&sender_id, &msg, MtmGetConnectedNodeMask(), &wait))
 	{
 		int			node_id;
 		const char *gid;

src/state.c

@@ -1196,8 +1196,9 @@ check_status_requests(MtmConfig *mtm_cfg)
 {
 	DmqSenderId sender_id;
 	StringInfoData msg;
+	bool		wait;
 
-	while(dmq_pop_nb(&sender_id, &msg, MtmGetConnectedNodeMask()))
+	while(dmq_pop_nb(&sender_id, &msg, MtmGetConnectedNodeMask(), &wait))
 	{
 		DmqDestinationId dest_id;
 		StringInfoData response_msg;