Don't send protocol messages to a shm_mq that no longer exists.

robertmhaas · robertmhaas · commit 26981d292758 · 2015-10-16T09:49:08.000-04:00
Commit 2bd9e41 introduced a mechanism for relaying protocol messages from a background worker to another backend via a shm_mq. However, there was no provision for shutting down the communication channel. Therefore, a protocol message sent late in the shutdown sequence, such as a DEBUG message resulting from cranking up log_min_messages, could crash the server. To fix, install an on_dsm_detach callback that disables sending messages to the shm_mq when the associated DSM is detached.
diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
@@ -867,7 +867,7 @@ ParallelWorkerMain(Datum main_arg)
 					 ParallelWorkerNumber * PARALLEL_ERROR_QUEUE_SIZE);
 	shm_mq_set_sender(mq, MyProc);
 	mqh = shm_mq_attach(mq, seg, NULL);
-	pq_redirect_to_shm_mq(mq, mqh);
+	pq_redirect_to_shm_mq(seg, mqh);
 	pq_set_parallel_master(fps->parallel_master_pid,
 						   fps->parallel_master_backend_id);
 
diff --git a/src/backend/libpq/pqmq.c b/src/backend/libpq/pqmq.c
@@ -26,6 +26,7 @@ static bool pq_mq_busy = false;
 static pid_t pq_mq_parallel_master_pid = 0;
 static pid_t pq_mq_parallel_master_backend_id = InvalidBackendId;
 
+static void pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg);
 static void mq_comm_reset(void);
 static int	mq_flush(void);
 static int	mq_flush_if_writable(void);
@@ -51,13 +52,26 @@ static PQcommMethods PqCommMqMethods = {
  * message queue.
  */
 void
-pq_redirect_to_shm_mq(shm_mq *mq, shm_mq_handle *mqh)
+pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
 {
 	PqCommMethods = &PqCommMqMethods;
-	pq_mq = mq;
+	pq_mq = shm_mq_get_queue(mqh);
 	pq_mq_handle = mqh;
 	whereToSendOutput = DestRemote;
 	FrontendProtocol = PG_PROTOCOL_LATEST;
+	on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
+}
+
+/*
+ * When the DSM that contains our shm_mq goes away, we need to stop sending
+ * messages to it.
+ */
+static void
+pq_cleanup_redirect_to_shm_mq(dsm_segment *seg, Datum arg)
+{
+	pq_mq = NULL;
+	pq_mq_handle = NULL;
+	whereToSendOutput = DestNone;
 }
 
 /*
@@ -123,9 +137,19 @@ mq_putmessage(char msgtype, const char *s, size_t len)
 		if (pq_mq != NULL)
 			shm_mq_detach(pq_mq);
 		pq_mq = NULL;
+		pq_mq_handle = NULL;
 		return EOF;
 	}
 
+	/*
+	 * If the message queue is already gone, just ignore the message. This
+	 * doesn't necessarily indicate a problem; for example, DEBUG messages
+	 * can be generated late in the shutdown sequence, after all DSMs have
+	 * already been detached.
+	 */
+	if (pq_mq == NULL)
+		return 0;
+
 	pq_mq_busy = true;
 
 	iov[0].data = &msgtype;
diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
@@ -745,6 +745,15 @@ shm_mq_detach(shm_mq *mq)
 		SetLatch(&victim->procLatch);
 }
 
+/*
+ * Get the shm_mq from handle.
+ */
+shm_mq *
+shm_mq_get_queue(shm_mq_handle *mqh)
+{
+	return mqh->mqh_queue;
+}
+
 /*
  * Write bytes into a shared message queue.
  */
diff --git a/src/include/libpq/pqmq.h b/src/include/libpq/pqmq.h
@@ -16,7 +16,7 @@
 #include "lib/stringinfo.h"
 #include "storage/shm_mq.h"
 
-extern void pq_redirect_to_shm_mq(shm_mq *, shm_mq_handle *);
+extern void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh);
 extern void pq_set_parallel_master(pid_t pid, BackendId backend_id);
 
 extern void pq_parse_errornotice(StringInfo str, ErrorData *edata);
diff --git a/src/include/storage/shm_mq.h b/src/include/storage/shm_mq.h
@@ -65,6 +65,9 @@ extern void shm_mq_set_handle(shm_mq_handle *, BackgroundWorkerHandle *);
 /* Break connection. */
 extern void shm_mq_detach(shm_mq *);
 
+/* Get the shm_mq from handle. */
+extern shm_mq *shm_mq_get_queue(shm_mq_handle *mqh);
+
 /* Send or receive messages. */
 extern shm_mq_result shm_mq_send(shm_mq_handle *mqh,
 			Size nbytes, const void *data, bool nowait);
