Bugfix on restart node problem.

OnNodeDisconnect callback added. DMQ receive error about instance disconnection and call the routine.
It is removes entry from hash table ExchShmem->htab referenced to disconnected node, make detach_receiver()
and destination_drop().
We do not needed in receiver_start callback because check (and establish, if needed) connections before
each query.

Author: danolivo

contrib/pg_exchange/dmq.c

@@ -1685,7 +1685,7 @@ dmq_get_destination_status(DmqDestinationId dest_id)
 }
 
 void
-dmq_destination_drop(char *receiver_name)
+dmq_destination_drop(const char *receiver_name)
 {
 	DmqDestinationId dest_id;
 	pid_t sender_pid;

contrib/pg_exchange/dmq.h

@@ -24,7 +24,7 @@ extern void dmq_init(const char *library_name);
 extern DmqDestinationId dmq_destination_add(char *connstr, char *sender_name,
 											char *receiver_name, int ping_period);
 extern DmqConnState dmq_get_destination_status(DmqDestinationId dest_id);
-extern void dmq_destination_drop(char *receiver_name);
+extern void dmq_destination_drop(const char *receiver_name);
 
 extern DmqSenderId dmq_attach_receiver(const char *sender_name, int mask_pos);
 extern void dmq_detach_receiver(const char *sender_name);
@@ -43,7 +43,7 @@ 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);
 
-typedef void (*dmq_receiver_hook_type) (char *);
+typedef void (*dmq_receiver_hook_type) (const char *);
 extern dmq_receiver_hook_type dmq_receiver_start_hook;
 extern dmq_receiver_hook_type dmq_receiver_stop_hook;
 

contrib/pg_exchange/pg_exchange.c

@@ -23,6 +23,7 @@
 #include "nodeDummyscan.h"
 #include "nodes/nodes.h"
 #include "utils/builtins.h"
+#include "utils/hsearch.h"
 #include "utils/guc.h"
 #include "utils/memutils.h"
 #include "utils/plancache.h"
@@ -37,10 +38,12 @@ PG_FUNCTION_INFO_V1(pg_exec_plan);
 
 #define DMQ_CONNSTR_MAX_LEN 150
 
+dmq_receiver_hook_type old_dmq_receiver_stop_hook;
 
 void _PG_init(void);
 static void deserialize_plan(char **squery, char **splan, char **sparams);
 static void exec_plan(char *squery, PlannedStmt *pstmt, ParamListInfo paramLI, const char *serverName);
+static void OnNodeDisconnect(const char *node_name);
 
 static Size
 shmem_size(void)
@@ -66,6 +69,9 @@ _PG_init(void)
 
 	RequestAddinShmemSpace(shmem_size());
 	RequestNamedLWLockTranche("pg_exchange", 1);
+
+	old_dmq_receiver_stop_hook = dmq_receiver_stop_hook;
+	dmq_receiver_stop_hook = OnNodeDisconnect;
 }
 
 Datum
@@ -198,3 +204,37 @@ exec_plan(char *squery, PlannedStmt *pstmt, ParamListInfo paramLI, const char *s
 	receiver->rDestroy(receiver);
 	ReleaseCachedPlan(cplan, false);
 }
+
+static void
+OnNodeDisconnect(const char *node_name)
+{
+	HASH_SEQ_STATUS status;
+	DMQDestinations *dest;
+	Oid serverid = InvalidOid;
+
+	elog(LOG, "Node %s: disconnected", node_name);
+
+
+	LWLockAcquire(ExchShmem->lock, LW_EXCLUSIVE);
+
+	hash_seq_init(&status, ExchShmem->htab);
+
+	while ((dest = hash_seq_search(&status)) != NULL)
+	{
+		if (!(strcmp(dest->node, node_name) == 0))
+			continue;
+
+		serverid = dest->serverid;
+		dmq_detach_receiver(node_name);
+		dmq_destination_drop(node_name);
+		break;
+	}
+	hash_seq_term(&status);
+
+	if (OidIsValid(serverid))
+		hash_search(ExchShmem->htab, &serverid, HASH_REMOVE, NULL);
+	else
+		elog(LOG, "Record on disconnected server %u with name %s not found.",
+														serverid, node_name);
+	LWLockRelease(ExchShmem->lock);
+}

contrib/pg_exchange/stream.c

@@ -227,14 +227,12 @@ ISendTuple(DmqDestinationId dest_id, char *stream, TupleTableSlot *slot,
 static void
 wait_for_delivery(OStream *ostream)
 {
-	int attempts;
-
-	for (attempts = 0; attempts < 50; attempts++)
+	for (;;)
 	{
 		int waits;
 
 		pg_usleep(10);
-		attempts++;
+
 		for (waits = 0; waits < 100000; waits++)
 		{
 			if (checkDelivery(ostream))

contrib/pg_execplan/tests/test.sql

@@ -0,0 +1,10 @@
+SELECT * FROM pt;
+SELECT * FROM rt;
+SELECT * FROM st;
+SELECT count(*) FROM pt;
+SELECT count(*) FROM pt,rt;
+SELECT count(*) FROM pt,rt,st;
+SELECT count(*) FROM pt,rt WHERE pt.id=rt.id;
+SELECT count(*) FROM pt,rt,st WHERE pt.id=rt.id and rt.id=st.id;
+SELECT count(*) FROM pt,rt,st WHERE pt.id=rt.id and rt.id=st.payload;
+SELECT count(*) FROM pt,rt,st WHERE pt.id=rt.payload and rt.id=st.payload;

contrib/pg_execplan/tests/test_restart.sh

@@ -0,0 +1,5 @@
+export PGDATABASE=test_base
+psql -p 5432 -c "SELECT * FROM pt;"
+pg_ctl -D PGDATA_n1 stop
+pg_ctl -w -c -o "-p 5433" -D PGDATA_n1 -l n1.log start
+psql -p 5432 -c "SELECT * FROM pt;"