Patch for stable execute at 3-node cluster

Author: danolivo

contrib/pg_exchange/Makefile

@@ -5,7 +5,8 @@ EXTENSION = pg_exchange
 EXTVERSION = 0.1
 PGFILEDESC = "pg_exchange - an exchange custom node and rules for the planner"
 MODULES = pg_exchange
-OBJS = pg_exchange.o exchange.o hooks.o common.o nodeDummyscan.o nodeDistPlanExec.o dmq.o stream.o partutils.o $(WIN32RES)
+OBJS =	common.o dmq.o exchange.o expath.o hooks.o nodeDistPlanExec.o \
+		nodeDummyscan.o partutils.o pg_exchange.o stream.o $(WIN32RES)
 
 fdw_srcdir = $(top_srcdir)/contrib/postgres_fdw/
 

contrib/pg_exchange/dmq.c

@@ -1239,27 +1239,44 @@ dmq_push(DmqDestinationId dest_id, char *stream_name, char *msg)
 	resetStringInfo(&buf);
 }
 
+static bool push_state = false;
+static StringInfoData buf;
 
-void
-dmq_push_buffer(DmqDestinationId dest_id, char *stream_name, const void *payload, size_t len)
+bool
+dmq_push_buffer(DmqDestinationId dest_id, char *stream_name,
+				const void *payload, size_t len, bool nowait)
 {
-	StringInfoData buf;
 	shm_mq_result res;
 
-	ensure_outq_handle();
+	if (!push_state)
+	{
+		ensure_outq_handle();
 
-	initStringInfo(&buf);
-	pq_sendbyte(&buf, dest_id);
-	pq_send_ascii_string(&buf, stream_name);
-	pq_sendbytes(&buf, payload, len);
+		initStringInfo(&buf);
+		pq_sendbyte(&buf, dest_id);
+		pq_send_ascii_string(&buf, stream_name);
+		pq_sendbytes(&buf, payload, len);
 
-	mtm_log(DmqTraceOutgoing, "[DMQ] pushing l=%d '%.*s'",
-			buf.len, buf.len, buf.data);
+		mtm_log(DmqTraceOutgoing, "[DMQ] pushing l=%d '%.*s'",
+				buf.len, buf.len, buf.data);
+	}
 
 	// XXX: use sendv instead
-	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, false);
+	res = shm_mq_send(dmq_local.mq_outh, buf.len, buf.data, nowait);
+
+	if (res == SHM_MQ_WOULD_BLOCK)
+	{
+		Assert(nowait == true);
+		push_state = true;
+		/* Report on full queue. */
+		return false;
+	}
+
 	if (res != SHM_MQ_SUCCESS)
 		mtm_log(WARNING, "[DMQ] dmq_push: can't send to queue");
+
+	push_state = false;
+	return true;
 }
 
 static bool
@@ -1467,6 +1484,17 @@ dmq_sender_name(DmqSenderId id)
 	return dmq_local.inhandles[id].name;
 }
 
+char *
+dmq_receiver_name(DmqDestinationId dest_id)
+{
+	char *recvName;
+
+	LWLockAcquire(dmq_state->lock, LW_SHARED);
+	recvName = pstrdup(dmq_state->destinations[dest_id].receiver_name);
+	LWLockRelease(dmq_state->lock);
+	return recvName;
+}
+
 DmqDestinationId
 dmq_remote_id(const char *name)
 {
@@ -1492,7 +1520,7 @@ dmq_remote_id(const char *name)
  * received. Returns false, if an error is occured.
  *
  * sender_id - identifier of the received message sender.
- * msg - buffer that contains received message.
+ * msg - pointer to local buffer that contains received message.
  * len - size of received message.
  */
 const char *

contrib/pg_exchange/dmq.h

@@ -15,7 +15,7 @@ typedef enum
 typedef int8 DmqDestinationId;
 typedef int8 DmqSenderId;
 
-#define DMQ_NAME_MAXLEN 32
+#define DMQ_NAME_MAXLEN 64
 #define DMQ_MAX_DESTINATIONS 127
 #define DMQ_MAX_RECEIVERS 100
 
@@ -38,6 +38,7 @@ extern void dmq_terminate_receiver(char *name);
 extern void dmq_stream_subscribe(const char *stream_name);
 extern void dmq_stream_unsubscribe(const char *stream_name);
 extern const char *dmq_sender_name(DmqSenderId id);
+extern char *dmq_receiver_name(DmqDestinationId dest_id);
 extern DmqDestinationId dmq_remote_id(const char *name);
 
 extern const char *
@@ -46,8 +47,8 @@ dmq_pop(DmqSenderId *sender_id, void **msg, Size *len, uint64 mask,
 extern bool dmq_pop_nb(DmqSenderId *sender_id, StringInfo msg, uint64 mask);
 
 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);
+extern bool dmq_push_buffer(DmqDestinationId dest_id, char *stream_name,
+							const void *buffer, size_t len, bool nowait);
 
 typedef void (*dmq_receiver_hook_type) (const char *);
 extern dmq_receiver_hook_type dmq_receiver_start_hook;