shared memory message queues

commit 1388d75ab847780ec9b208787724029ed3795520
Author: Robert Haas <rhaas@postgresql.org>
Date:   Tue Oct 1 11:39:52 2013 -0400

    Allow on-detach callbacks for shared memory segments.
    
    Just as backends must clean up their shared memory state (releasing
    lwlocks, buffer pins, etc.) before exiting, they must also perform
    any similar cleanups related to dynamic shared memory segments they
    have mapped.  So add a mechanism to ensure that.
    
    Existing on_shmem_exit hooks include both "user level" cleanup such
    as transaction abort and removal of leftover temporary relations and
    also "low level" cleanup that forcibly released leftover shared
    memory resources.  On-detach callbacks should run after the first
    group but before the second group, so split on_shmem_exit into early
    and late callbacks for that purpose.

diff --git a/contrib/pg_stat_statements/pg_stat_statements.c b/contrib/pg_stat_statements/pg_stat_statements.c
index ea930af..770f499 100644
--- a/contrib/pg_stat_statements/pg_stat_statements.c
+++ b/contrib/pg_stat_statements/pg_stat_statements.c
@@ -439,7 +439,7 @@ pgss_shmem_startup(void)
 	 * exit hook to dump the statistics to disk.
 	 */
 	if (!IsUnderPostmaster)
-		on_shmem_exit(pgss_shmem_shutdown, (Datum) 0);
+		on_shmem_exit(pgss_shmem_shutdown, (Datum) 0, SHMEM_EXIT_LATE);
 
 	/*
 	 * Attempt to load old statistics from the dump file, if this is the first
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 62f4c18..4c41f8a 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -403,7 +403,7 @@ AuxiliaryProcessMain(int argc, char *argv[])
 		InitBufferPoolBackend();
 
 		/* register a shutdown callback for LWLock cleanup */
-		on_shmem_exit(ShutdownAuxiliaryProcess, 0);
+		on_shmem_exit(ShutdownAuxiliaryProcess, 0, SHMEM_EXIT_EARLY);
 	}
 
 	/*
diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 4434dd6..3b4e0ca 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -3681,7 +3681,7 @@ AtEOXact_Namespace(bool isCommit)
 	if (myTempNamespaceSubID != InvalidSubTransactionId)
 	{
 		if (isCommit)
-			on_shmem_exit(RemoveTempRelationsCallback, 0);
+			on_shmem_exit(RemoveTempRelationsCallback, 0, SHMEM_EXIT_EARLY);
 		else
 		{
 			myTempNamespace = InvalidOid;
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 6414291..f273c0f 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -921,7 +921,7 @@ Exec_ListenPreCommit(void)
 	 */
 	if (!unlistenExitRegistered)
 	{
-		on_shmem_exit(Async_UnlistenOnExit, 0);
+		on_shmem_exit(Async_UnlistenOnExit, 0, SHMEM_EXIT_EARLY);
 		unlistenExitRegistered = true;
 	}
 
diff --git a/src/backend/port/ipc_test.c b/src/backend/port/ipc_test.c
index 6020ef5..2e067ea 100644
--- a/src/backend/port/ipc_test.c
+++ b/src/backend/port/ipc_test.c
@@ -56,14 +56,19 @@ char	   *DataDir = ".";
 
 #define MAX_ON_EXITS 20
 
-static struct ONEXIT
+struct ONEXIT
 {
 	pg_on_exit_callback function;
 	Datum		arg;
-}	on_proc_exit_list[MAX_ON_EXITS], on_shmem_exit_list[MAX_ON_EXITS];
+};
+
+static struct ONEXIT on_proc_exit_list[MAX_ON_EXITS];
+static struct ONEXIT on_shmem_exit_early_list[MAX_ON_EXITS];
+static struct ONEXIT on_shmem_exit_late_list[MAX_ON_EXITS];
 
 static int	on_proc_exit_index,
-			on_shmem_exit_index;
+			on_shmem_exit_early_index,
+			on_shmem_exit_late_index;
 
 void
 proc_exit(int code)
@@ -78,28 +83,51 @@ proc_exit(int code)
 void
 shmem_exit(int code)
 {
-	while (--on_shmem_exit_index >= 0)
-		(*on_shmem_exit_list[on_shmem_exit_index].function) (code,
-								on_shmem_exit_list[on_shmem_exit_index].arg);
-	on_shmem_exit_index = 0;
+	while (--on_shmem_exit_early_index >= 0)
+		(*on_shmem_exit_early_list[on_shmem_exit_early_index].function) (code,
+					on_shmem_exit_early_list[on_shmem_exit_early_index].arg);
+	on_shmem_exit_early_index = 0;
+
+	while (--on_shmem_exit_late_index >= 0)
+		(*on_shmem_exit_late_list[on_shmem_exit_late_index].function) (code,
+					on_shmem_exit_late_list[on_shmem_exit_late_index].arg);
+	on_shmem_exit_late_index = 0;
 }
 
 void
-on_shmem_exit(pg_on_exit_callback function, Datum arg)
+on_shmem_exit(pg_on_exit_callback function, Datum arg, shmem_exit_phase phase)
 {
-	if (on_shmem_exit_index >= MAX_ON_EXITS)
-		elog(FATAL, "out of on_shmem_exit slots");
+	int	   *index;
+	struct ONEXIT *list;
+
+	if (phase == SHMEM_EXIT_EARLY)
+	{
+		index = &on_shmem_exit_early_index;
+		list = on_shmem_exit_early_list;
+	}
+	else
+	{
+		Assert(phase == SHMEM_EXIT_LATE);
+		index = &on_shmem_exit_late_index;
+		list = on_shmem_exit_late_list;
+	}
+
+	if (*index >= MAX_ON_EXITS)
+		ereport(FATAL,
+				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+				 errmsg_internal("out of on_shmem_exit slots")));
 
-	on_shmem_exit_list[on_shmem_exit_index].function = function;
-	on_shmem_exit_list[on_shmem_exit_index].arg = arg;
+	list[*index].function = function;
+	list[*index].arg = arg;
 
-	++on_shmem_exit_index;
+	++*index;
 }
 
 void
 on_exit_reset(void)
 {
-	on_shmem_exit_index = 0;
+	on_shmem_exit_early_index = 0;
+	on_shmem_exit_late_index = 0;
 	on_proc_exit_index = 0;
 }
 
diff --git a/src/backend/port/posix_sema.c b/src/backend/port/posix_sema.c
index 061fd2d..ff0bb64 100644
--- a/src/backend/port/posix_sema.c
+++ b/src/backend/port/posix_sema.c
@@ -160,7 +160,7 @@ PGReserveSemaphores(int maxSemas, int port)
 	maxSems = maxSemas;
 	nextSemKey = port * 1000;
 
-	on_shmem_exit(ReleaseSemaphores, 0);
+	on_shmem_exit(ReleaseSemaphores, 0, SHMEM_EXIT_LATE);
 }
 
 /*
diff --git a/src/backend/port/sysv_sema.c b/src/backend/port/sysv_sema.c
index 2988561..2e10934 100644
--- a/src/backend/port/sysv_sema.c
+++ b/src/backend/port/sysv_sema.c
@@ -296,7 +296,7 @@ PGReserveSemaphores(int maxSemas, int port)
 	nextSemaKey = port * 1000;
 	nextSemaNumber = SEMAS_PER_SET;		/* force sema set alloc on 1st call */
 
-	on_shmem_exit(ReleaseSemaphores, 0);
+	on_shmem_exit(ReleaseSemaphores, 0, SHMEM_EXIT_LATE);
 }
 
 /*
diff --git a/src/backend/port/sysv_shmem.c b/src/backend/port/sysv_shmem.c
index b604407..aa80765 100644
--- a/src/backend/port/sysv_shmem.c
+++ b/src/backend/port/sysv_shmem.c
@@ -165,7 +165,7 @@ InternalIpcMemoryCreate(IpcMemoryKey memKey, Size size)
 	}
 
 	/* Register on-exit routine to delete the new segment */
-	on_shmem_exit(IpcMemoryDelete, Int32GetDatum(shmid));
+	on_shmem_exit(IpcMemoryDelete, Int32GetDatum(shmid), SHMEM_EXIT_LATE);
 
 	/* OK, should be able to attach to the segment */
 	memAddress = shmat(shmid, NULL, PG_SHMAT_FLAGS);
@@ -174,7 +174,8 @@ InternalIpcMemoryCreate(IpcMemoryKey memKey, Size size)
 		elog(FATAL, "shmat(id=%d) failed: %m", shmid);
 
 	/* Register on-exit routine to detach new segment before deleting */
-	on_shmem_exit(IpcMemoryDetach, PointerGetDatum(memAddress));
+	on_shmem_exit(IpcMemoryDetach, PointerGetDatum(memAddress),
+				  SHMEM_EXIT_LATE);
 
 	/*
 	 * Store shmem key and ID in data directory lockfile.  Format to try to
diff --git a/src/backend/port/win32_sema.c b/src/backend/port/win32_sema.c
index dc5054b..3f02fea 100644
--- a/src/backend/port/win32_sema.c
+++ b/src/backend/port/win32_sema.c
@@ -41,7 +41,7 @@ PGReserveSemaphores(int maxSemas, int port)
 	numSems = 0;
 	maxSems = maxSemas;
 
-	on_shmem_exit(ReleaseSemaphores, 0);
+	on_shmem_exit(ReleaseSemaphores, 0, SHMEM_EXIT_LATE);
 }
 
 /*
diff --git a/src/backend/port/win32_shmem.c b/src/backend/port/win32_shmem.c
index 0db8e8f..28cb739 100644
--- a/src/backend/port/win32_shmem.c
+++ b/src/backend/port/win32_shmem.c
@@ -212,7 +212,8 @@ PGSharedMemoryCreate(Size size, bool makePrivate, int port)
 
 
 	/* Register on-exit routine to delete the new segment */
-	on_shmem_exit(pgwin32_SharedMemoryDelete, PointerGetDatum(hmap2));
+	on_shmem_exit(pgwin32_SharedMemoryDelete, PointerGetDatum(hmap2),
+				  SHMEM_EXIT_LATE);
 
 	/*
 	 * Get a pointer to the new shared memory segment. Map the whole segment
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 8c14d0f..6596fd3 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -1624,7 +1624,7 @@ AutoVacWorkerMain(int argc, char *argv[])
 		AutoVacuumShmem->av_startingWorker = NULL;
 		LWLockRelease(AutovacuumLock);
 
-		on_shmem_exit(FreeWorkerInfo, 0);
+		on_shmem_exit(FreeWorkerInfo, 0, SHMEM_EXIT_LATE);
 
 		/* wake up the launcher */
 		if (AutoVacuumShmem->av_launcherpid != 0)
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index b5ce2f6..2e2351e 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -2459,7 +2459,7 @@ pgstat_initialize(void)
 	MyBEEntry = &BackendStatusArray[MyBackendId - 1];
 
 	/* Set up a process-exit hook to clean up */
-	on_shmem_exit(pgstat_beshutdown_hook, 0);
+	on_shmem_exit(pgstat_beshutdown_hook, 0, SHMEM_EXIT_LATE);
 }
 
 /* ----------
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 413f0b9..ea1773f 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -250,7 +250,7 @@ WalReceiverMain(void)
 	SpinLockRelease(&walrcv->mutex);
 
 	/* Arrange to clean up at walreceiver exit */
-	on_shmem_exit(WalRcvDie, 0);
+	on_shmem_exit(WalRcvDie, 0, SHMEM_EXIT_LATE);
 
 	OwnLatch(&walrcv->latch);
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index afd559d..a4505c2 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1207,7 +1207,7 @@ InitWalSenderSlot(void)
 						max_wal_senders)));
 
 	/* Arrange to clean up at walsender exit */
-	on_shmem_exit(WalSndKill, 0);
+	on_shmem_exit(WalSndKill, 0, SHMEM_EXIT_LATE);
 }
 
 /* Destroy the per-walsender data structure for this walsender process */
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index f848391..276e29e 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -1741,7 +1741,7 @@ AtEOXact_Buffers(bool isCommit)
 void
 InitBufferPoolBackend(void)
 {
-	on_shmem_exit(AtProcExit_Buffers, 0);
+	on_shmem_exit(AtProcExit_Buffers, 0, SHMEM_EXIT_LATE);
 }
 
 /*
diff --git a/src/backend/storage/ipc/dsm.c b/src/backend/storage/ipc/dsm.c
index 9f9b91a..decf203 100644
--- a/src/backend/storage/ipc/dsm.c
+++ b/src/backend/storage/ipc/dsm.c
@@ -58,6 +58,14 @@
 
 #define INVALID_CONTROL_SLOT		((uint32) -1)
 
+/* Backend-local tracking for on-detach callbacks. */
+typedef struct dsm_segment_detach_callback
+{
+	on_dsm_detach_callback	function;
+	Datum					arg;
+	slist_node				node;
+} dsm_segment_detach_callback;
+
 /* Backend-local state for a dynamic shared memory segment. */
 struct dsm_segment
 {
@@ -68,6 +76,7 @@ struct dsm_segment
 	void       *impl_private;		/* Implementation-specific private data. */
 	void	   *mapped_address;		/* Mapping address, or NULL if unmapped. */
 	Size		mapped_size;		/* Size of our mapping. */
+	slist_head	on_detach;			/* On-detach callbacks. */
 };
 
 /* Shared-memory state for a dynamic shared memory segment. */
@@ -91,7 +100,6 @@ static void dsm_cleanup_for_mmap(void);
 static bool dsm_read_state_file(dsm_handle *h);
 static void dsm_write_state_file(dsm_handle h);
 static void dsm_postmaster_shutdown(int code, Datum arg);
-static void dsm_backend_shutdown(int code, Datum arg);
 static dsm_segment *dsm_create_descriptor(void);
 static bool dsm_control_segment_sane(dsm_control_header *control,
 						 Size mapped_size);
@@ -191,7 +199,7 @@ dsm_postmaster_startup(void)
 			break;
 	}
 	dsm_control = dsm_control_address;
-	on_shmem_exit(dsm_postmaster_shutdown, 0);
+	on_shmem_exit(dsm_postmaster_shutdown, 0, SHMEM_EXIT_LATE);
 	elog(DEBUG2,
 		 "created dynamic shared memory control segment %u (%lu bytes)",
 		 dsm_control_handle, (unsigned long) segsize);
@@ -553,9 +561,6 @@ dsm_backend_startup(void)
 	}
 #endif
 
-	/* Arrange to detach segments on exit. */
-	on_shmem_exit(dsm_backend_shutdown, 0);
-
 	dsm_init_done = true;
 }
 
@@ -715,8 +720,8 @@ dsm_attach(dsm_handle h)
 /*
  * At backend shutdown time, detach any segments that are still attached.
  */
-static void
-dsm_backend_shutdown(int code, Datum arg)
+void
+dsm_backend_shutdown(void)
 {
 	while (!dlist_is_empty(&dsm_segment_list))
 	{
@@ -772,6 +777,27 @@ void
 dsm_detach(dsm_segment *seg)
 {
 	/*
+	 * Invoke registered callbacks.  Just in case one of those callbacks
+	 * throws a further error that brings us back here, pop the callback
+	 * before invoking it, to avoid infinite error recursion.
+	 */
+	while (!slist_is_empty(&seg->on_detach))
+	{
+		slist_node *node;
+		dsm_segment_detach_callback *cb;
+		on_dsm_detach_callback	function;
+		Datum		arg;
+
+		node = slist_pop_head_node(&seg->on_detach);
+		cb = slist_container(dsm_segment_detach_callback, node, node);
+		function = cb->function;
+		arg = cb->arg;
+		pfree(cb);
+
+		function(seg, arg);
+	}
+
+	/*
 	 * Try to remove the mapping, if one exists.  Normally, there will be,
 	 * but maybe not, if we failed partway through a create or attach
 	 * operation.  We remove the mapping before decrementing the reference
@@ -913,6 +939,44 @@ dsm_segment_handle(dsm_segment *seg)
 }
 
 /*
+ * Register an on-detach callback for a dynamic shared memory segment.
+ */
+void
+on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
+{
+	dsm_segment_detach_callback *cb;
+
+	cb = MemoryContextAlloc(TopMemoryContext,
+							sizeof(dsm_segment_detach_callback));
+	cb->function = function;
+	cb->arg = arg;
+	slist_push_head(&seg->on_detach, &cb->node);
+}
+
+/*
+ * Unregister an on-detach callback for a dynamic shared memory segment.
+ */
+void
+on_dsm_detach_cancel(dsm_segment *seg, on_dsm_detach_callback function,
+					 Datum arg)
+{
+	slist_mutable_iter	iter;
+
+	slist_foreach_modify(iter, &seg->on_detach)
+	{
+		dsm_segment_detach_callback *cb;
+
+		cb = slist_container(dsm_segment_detach_callback, node, iter.cur);
+		if (cb->function == function && cb->arg == arg)
+		{
+			slist_delete_current(&iter);
+			pfree(cb);
+			break;
+		}
+	}
+}
+
+/*
  * Create a segment descriptor.
  */
 static dsm_segment *
@@ -934,6 +998,8 @@ dsm_create_descriptor(void)
 	seg->resowner = CurrentResourceOwner;
 	ResourceOwnerRememberDSM(CurrentResourceOwner, seg);
 
+	slist_init(&seg->on_detach);
+
 	return seg;
 }
 
diff --git a/src/backend/storage/ipc/ipc.c b/src/backend/storage/ipc/ipc.c
index c339e9c..ae42366 100644
--- a/src/backend/storage/ipc/ipc.c
+++ b/src/backend/storage/ipc/ipc.c
@@ -27,6 +27,7 @@
 #ifdef PROFILE_PID_DIR
 #include "postmaster/autovacuum.h"
 #endif
+#include "storage/dsm.h"
 #include "storage/ipc.h"
 #include "tcop/tcopprot.h"
 
@@ -64,14 +65,19 @@ static void proc_exit_prepare(int code);
 
 #define MAX_ON_EXITS 20
 
-static struct ONEXIT
+struct ONEXIT
 {
 	pg_on_exit_callback function;
 	Datum		arg;
-}	on_proc_exit_list[MAX_ON_EXITS], on_shmem_exit_list[MAX_ON_EXITS];
+};
+
+static struct ONEXIT on_proc_exit_list[MAX_ON_EXITS];
+static struct ONEXIT on_shmem_exit_early_list[MAX_ON_EXITS];
+static struct ONEXIT on_shmem_exit_late_list[MAX_ON_EXITS];
 
 static int	on_proc_exit_index,
-			on_shmem_exit_index;
+			on_shmem_exit_early_index,
+			on_shmem_exit_late_index;
 
 
 /* ----------------------------------------------------------------
@@ -202,26 +208,61 @@ proc_exit_prepare(int code)
 /* ------------------
  * Run all of the on_shmem_exit routines --- but don't actually exit.
  * This is used by the postmaster to re-initialize shared memory and
- * semaphores after a backend dies horribly.
+ * semaphores after a backend dies horribly.  As with proc_exit(), we
+ * remove each callback from the list before calling it, to avoid
+ * infinite loop in case of error.
  * ------------------
  */
 void
 shmem_exit(int code)
 {
-	elog(DEBUG3, "shmem_exit(%d): %d callbacks to make",
-		 code, on_shmem_exit_index);
+	/*
+	 * Call early callbacks.
+	 *
+	 * These are generally things that need most of the system to still be
+	 * up and working, such as cleanup of temp relations, which requires
+	 * catalog access; or things that need to be completed because later
+	 * cleanup steps depend on them, such as releasing lwlocks.
+	 */
+	elog(DEBUG3, "shmem_exit(%d): %d early callbacks to make",
+		 code, on_shmem_exit_early_index);
+	while (--on_shmem_exit_early_index >= 0)
+		(*on_shmem_exit_early_list[on_shmem_exit_early_index].function) (code,
+					on_shmem_exit_early_list[on_shmem_exit_early_index].arg);
+	on_shmem_exit_early_index = 0;
 
 	/*
-	 * call all the registered callbacks.
+	 * Call dynamic shared memory callbacks.
+	 *
+	 * These serve the same purpose as late callbacks, but for dynamic shared
+	 * memory segments rather than the main shared memory segment.
+	 * dsm_backend_shutdown() has the same kind of progressive logic we use
+	 * for the main shared memory segment; namely, it unregisters each
+	 * callback before invoking it, so that we don't get stuck in an infinite
+	 * loop if one of those callbacks itself throws an ERROR or FATAL.
 	 *
-	 * As with proc_exit(), we remove each callback from the list before
-	 * calling it, to avoid infinite loop in case of error.
+	 * Note that explicitly calling this function here is quite different
+	 * from registering it as an on_shmem_exit callback for precisely this
+	 * reason: if one dynamic shared memory callback errors out, the remaining
+	 * callbacks will still be invoked.  Thus, hard-coding this call puts it
+	 * equal footing with callbacks for the main shared memory segment.
 	 */
-	while (--on_shmem_exit_index >= 0)
-		(*on_shmem_exit_list[on_shmem_exit_index].function) (code,
-								on_shmem_exit_list[on_shmem_exit_index].arg);
+	dsm_backend_shutdown();
 
-	on_shmem_exit_index = 0;
+	/*
+	 * Call late callbacks.
+	 *
+	 * These are generally releasing low-level shared memory resources.  In
+	 * some cases, this is a backstop against the possibility that the early
+	 * callbacks might themselves fail, leading to re-entry to this routine;
+	 * in other cases, it's cleanup that only happens at process exit.
+	 */
+	elog(DEBUG3, "shmem_exit(%d): %d late callbacks to make",
+		 code, on_shmem_exit_late_index);
+	while (--on_shmem_exit_late_index >= 0)
+		(*on_shmem_exit_late_list[on_shmem_exit_late_index].function) (code,
+					on_shmem_exit_late_list[on_shmem_exit_late_index].arg);
+	on_shmem_exit_late_index = 0;
 }
 
 /* ----------------------------------------------------------------
@@ -277,17 +318,32 @@ on_proc_exit(pg_on_exit_callback function, Datum arg)
  * ----------------------------------------------------------------
  */
 void
-on_shmem_exit(pg_on_exit_callback function, Datum arg)
+on_shmem_exit(pg_on_exit_callback function, Datum arg, shmem_exit_phase phase)
 {
-	if (on_shmem_exit_index >= MAX_ON_EXITS)
+	int	   *index;
+	struct ONEXIT *list;
+
+	if (phase == SHMEM_EXIT_EARLY)
+	{
+		index = &on_shmem_exit_early_index;
+		list = on_shmem_exit_early_list;
+	}
+	else
+	{
+		Assert(phase == SHMEM_EXIT_LATE);
+		index = &on_shmem_exit_late_index;
+		list = on_shmem_exit_late_list;
+	}
+
+	if (*index >= MAX_ON_EXITS)
 		ereport(FATAL,
 				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
 				 errmsg_internal("out of on_shmem_exit slots")));
 
-	on_shmem_exit_list[on_shmem_exit_index].function = function;
-	on_shmem_exit_list[on_shmem_exit_index].arg = arg;
+	list[*index].function = function;
+	list[*index].arg = arg;
 
-	++on_shmem_exit_index;
+	++*index;
 
 	if (!atexit_callback_setup)
 	{
@@ -306,12 +362,28 @@ on_shmem_exit(pg_on_exit_callback function, Datum arg)
  * ----------------------------------------------------------------
  */
 void
-cancel_shmem_exit(pg_on_exit_callback function, Datum arg)
+cancel_shmem_exit(pg_on_exit_callback function, Datum arg,
+				  shmem_exit_phase phase)
 {
-	if (on_shmem_exit_index > 0 &&
-		on_shmem_exit_list[on_shmem_exit_index - 1].function == function &&
-		on_shmem_exit_list[on_shmem_exit_index - 1].arg == arg)
-		--on_shmem_exit_index;
+	int	   *index;
+	struct ONEXIT *list;
+
+	if (phase == SHMEM_EXIT_EARLY)
+	{
+		index = &on_shmem_exit_early_index;
+		list = on_shmem_exit_early_list;
+	}
+	else
+	{
+		Assert(phase == SHMEM_EXIT_LATE);
+		index = &on_shmem_exit_late_index;
+		list = on_shmem_exit_late_list;
+	}
+
+	if (*index > 0 &&
+		list[*index - 1].function == function &&
+		list[*index - 1].arg == arg)
+		--*index;
 }
 
 /* ----------------------------------------------------------------
@@ -326,6 +398,7 @@ cancel_shmem_exit(pg_on_exit_callback function, Datum arg)
 void
 on_exit_reset(void)
 {
-	on_shmem_exit_index = 0;
+	on_shmem_exit_early_index = 0;
+	on_shmem_exit_late_index = 0;
 	on_proc_exit_index = 0;
 }
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index c4b5d01..2533bf1 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -131,7 +131,8 @@ ProcSignalInit(int pss_idx)
 	MyProcSignalSlot = slot;
 
 	/* Set up to release the slot on process exit */
-	on_shmem_exit(CleanupProcSignalState, Int32GetDatum(pss_idx));
+	on_shmem_exit(CleanupProcSignalState, Int32GetDatum(pss_idx),
+				  SHMEM_EXIT_LATE);
 }
 
 /*
diff --git a/src/backend/storage/ipc/sinvaladt.c b/src/backend/storage/ipc/sinvaladt.c
index 44d02c5..aaf958a 100644
--- a/src/backend/storage/ipc/sinvaladt.c
+++ b/src/backend/storage/ipc/sinvaladt.c
@@ -325,7 +325,8 @@ SharedInvalBackendInit(bool sendOnly)
 	LWLockRelease(SInvalWriteLock);
 
 	/* register exit routine to mark my entry inactive at exit */
-	on_shmem_exit(CleanupInvalidationState, PointerGetDatum(segP));
+	on_shmem_exit(CleanupInvalidationState, PointerGetDatum(segP),
+				  SHMEM_EXIT_LATE);
 
 	elog(DEBUG4, "my backend ID is %d", MyBackendId);
 }
diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c
index 4f88d3f..b1af5c6 100644
--- a/src/backend/storage/lmgr/lwlock.c
+++ b/src/backend/storage/lmgr/lwlock.c
@@ -138,7 +138,7 @@ init_lwlock_stats(void)
 	spin_delay_counts = calloc(numLocks, sizeof(int));
 	block_counts = calloc(numLocks, sizeof(int));
 	counts_for_pid = MyProcPid;
-	on_shmem_exit(print_lwlock_stats, 0);
+	on_shmem_exit(print_lwlock_stats, 0, SHMEM_EXIT_LATE);
 }
 
 static void
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
index 222251d..8dffea9 100644
--- a/src/backend/storage/lmgr/proc.c
+++ b/src/backend/storage/lmgr/proc.c
@@ -406,7 +406,7 @@ InitProcess(void)
 	/*
 	 * Arrange to clean up at backend exit.
 	 */
-	on_shmem_exit(ProcKill, 0);
+	on_shmem_exit(ProcKill, 0, SHMEM_EXIT_LATE);
 
 	/*
 	 * Now that we have a PGPROC, we could try to acquire locks, so initialize
@@ -435,7 +435,7 @@ InitProcessPhase2(void)
 	/*
 	 * Arrange to clean that up at backend exit.
 	 */
-	on_shmem_exit(RemoveProcFromArray, 0);
+	on_shmem_exit(RemoveProcFromArray, 0, SHMEM_EXIT_LATE);
 }
 
 /*
@@ -563,7 +563,8 @@ InitAuxiliaryProcess(void)
 	/*
 	 * Arrange to clean up at process exit.
 	 */
-	on_shmem_exit(AuxiliaryProcKill, Int32GetDatum(proctype));
+	on_shmem_exit(AuxiliaryProcKill, Int32GetDatum(proctype),
+				  SHMEM_EXIT_LATE);
 }
 
 /*
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 2c7f0f1..97a7d49 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -560,7 +560,7 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
 		 * down at exit.
 		 */
 		StartupXLOG();
-		on_shmem_exit(ShutdownXLOG, 0);
+		on_shmem_exit(ShutdownXLOG, 0, SHMEM_EXIT_LATE);
 	}
 
 	/*
@@ -587,15 +587,14 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
 	RelationCacheInitializePhase2();
 
 	/*
-	 * Set up process-exit callback to do pre-shutdown cleanup.  This has to
-	 * be after we've initialized all the low-level modules like the buffer
-	 * manager, because during shutdown this has to run before the low-level
-	 * modules start to close down.  On the other hand, we want it in place
-	 * before we begin our first transaction --- if we fail during the
-	 * initialization transaction, as is entirely possible, we need the
-	 * AbortTransaction call to clean up.
+	 * Set up process-exit callback to do pre-shutdown cleanup.  This is the
+	 * first callback we register as SHMEM_EXIT_EARLY; thus, this will be the
+	 * last thing we do before low-level modules like the buffer manager begin
+	 * to close down.  We need to have this in place before we begin our first
+	 * transaction --- if we fail during the initialization transaction, as is
+	 * entirely possible, we need the AbortTransaction call to clean up.
 	 */
-	on_shmem_exit(ShutdownPostgres, 0);
+	on_shmem_exit(ShutdownPostgres, 0, SHMEM_EXIT_EARLY);
 
 	/* The autovacuum launcher is done here */
 	if (IsAutoVacuumLauncherProcess())
diff --git a/src/include/storage/dsm.h b/src/include/storage/dsm.h
index d906eba..5c7c981 100644
--- a/src/include/storage/dsm.h
+++ b/src/include/storage/dsm.h
@@ -17,8 +17,9 @@
 
 typedef struct dsm_segment dsm_segment;
 
-/* Initialization function. */
+/* Startup and shutdown functions. */
 extern void dsm_postmaster_startup(void);
+extern void dsm_backend_shutdown(void);
 
 /* Functions that create, update, or remove mappings. */
 extern dsm_segment *dsm_create(Size size);
@@ -36,4 +37,11 @@ extern void *dsm_segment_address(dsm_segment *seg);
 extern Size dsm_segment_map_length(dsm_segment *seg);
 extern dsm_handle dsm_segment_handle(dsm_segment *seg);
 
+/* Cleanup hooks. */
+typedef void (*on_dsm_detach_callback) (dsm_segment *, Datum arg);
+extern void on_dsm_detach(dsm_segment *seg,
+			  on_dsm_detach_callback function, Datum arg);
+extern void on_dsm_detach_cancel(dsm_segment *seg,
+			  on_dsm_detach_callback function, Datum arg);
+
 #endif   /* DSM_H */
diff --git a/src/include/storage/ipc.h b/src/include/storage/ipc.h
index ac4ac66..4989abb 100644
--- a/src/include/storage/ipc.h
+++ b/src/include/storage/ipc.h
@@ -21,6 +21,12 @@
 typedef void (*pg_on_exit_callback) (int code, Datum arg);
 typedef void (*shmem_startup_hook_type) (void);
 
+typedef enum
+{
+	SHMEM_EXIT_EARLY,		/* user-level cleanup and transaction abort */
+	SHMEM_EXIT_LATE			/* low-level subsystem shutdown */
+} shmem_exit_phase;
+
 /*----------
  * API for handling cleanup that must occur during either ereport(ERROR)
  * or ereport(FATAL) exits from a block of code.  (Typical examples are
@@ -46,14 +52,14 @@ typedef void (*shmem_startup_hook_type) (void);
  */
 #define PG_ENSURE_ERROR_CLEANUP(cleanup_function, arg)	\
 	do { \
-		on_shmem_exit(cleanup_function, arg); \
+		on_shmem_exit(cleanup_function, arg, SHMEM_EXIT_EARLY); \
 		PG_TRY()
 
 #define PG_END_ENSURE_ERROR_CLEANUP(cleanup_function, arg)	\
-		cancel_shmem_exit(cleanup_function, arg); \
+		cancel_shmem_exit(cleanup_function, arg, SHMEM_EXIT_EARLY); \
 		PG_CATCH(); \
 		{ \
-			cancel_shmem_exit(cleanup_function, arg); \
+			cancel_shmem_exit(cleanup_function, arg, SHMEM_EXIT_EARLY); \
 			cleanup_function (0, arg); \
 			PG_RE_THROW(); \
 		} \
@@ -67,8 +73,10 @@ extern bool proc_exit_inprogress;
 extern void proc_exit(int code) __attribute__((noreturn));
 extern void shmem_exit(int code);
 extern void on_proc_exit(pg_on_exit_callback function, Datum arg);
-extern void on_shmem_exit(pg_on_exit_callback function, Datum arg);
-extern void cancel_shmem_exit(pg_on_exit_callback function, Datum arg);
+extern void on_shmem_exit(pg_on_exit_callback function, Datum arg,
+			  shmem_exit_phase phase);
+extern void cancel_shmem_exit(pg_on_exit_callback function, Datum arg,
+				  shmem_exit_phase phase);
 extern void on_exit_reset(void);
 
 /* ipci.c */

diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 62f4c18..938cf7a 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -402,8 +402,8 @@ AuxiliaryProcessMain(int argc, char *argv[])
 		/* finish setting up bufmgr.c */
 		InitBufferPoolBackend();
 
-		/* register a shutdown callback for LWLock cleanup */
-		on_shmem_exit(ShutdownAuxiliaryProcess, 0);
+		/* register a before-shutdown callback for LWLock cleanup */
+		before_shmem_exit(ShutdownAuxiliaryProcess, 0);
 	}
 
 	/*
diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 4434dd6..2f78e37 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -3681,7 +3681,7 @@ AtEOXact_Namespace(bool isCommit)
 	if (myTempNamespaceSubID != InvalidSubTransactionId)
 	{
 		if (isCommit)
-			on_shmem_exit(RemoveTempRelationsCallback, 0);
+			before_shmem_exit(RemoveTempRelationsCallback, 0);
 		else
 		{
 			myTempNamespace = InvalidOid;
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index f4526a6..5311e1f 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -921,7 +921,7 @@ Exec_ListenPreCommit(void)
 	 */
 	if (!unlistenExitRegistered)
 	{
-		on_shmem_exit(Async_UnlistenOnExit, 0);
+		before_shmem_exit(Async_UnlistenOnExit, 0);
 		unlistenExitRegistered = true;
 	}
 
diff --git a/src/backend/storage/ipc/dsm.c b/src/backend/storage/ipc/dsm.c
index a33700e..19eb878 100644
--- a/src/backend/storage/ipc/dsm.c
+++ b/src/backend/storage/ipc/dsm.c
@@ -58,6 +58,14 @@
 
 #define INVALID_CONTROL_SLOT		((uint32) -1)
 
+/* Backend-local tracking for on-detach callbacks. */
+typedef struct dsm_segment_detach_callback
+{
+	on_dsm_detach_callback	function;
+	Datum					arg;
+	slist_node				node;
+} dsm_segment_detach_callback;
+
 /* Backend-local state for a dynamic shared memory segment. */
 struct dsm_segment
 {
@@ -68,6 +76,7 @@ struct dsm_segment
 	void       *impl_private;		/* Implementation-specific private data. */
 	void	   *mapped_address;		/* Mapping address, or NULL if unmapped. */
 	Size		mapped_size;		/* Size of our mapping. */
+	slist_head	on_detach;			/* On-detach callbacks. */
 };
 
 /* Shared-memory state for a dynamic shared memory segment. */
@@ -91,7 +100,6 @@ static void dsm_cleanup_for_mmap(void);
 static bool dsm_read_state_file(dsm_handle *h);
 static void dsm_write_state_file(dsm_handle h);
 static void dsm_postmaster_shutdown(int code, Datum arg);
-static void dsm_backend_shutdown(int code, Datum arg);
 static dsm_segment *dsm_create_descriptor(void);
 static bool dsm_control_segment_sane(dsm_control_header *control,
 						 Size mapped_size);
@@ -556,9 +564,6 @@ dsm_backend_startup(void)
 	}
 #endif
 
-	/* Arrange to detach segments on exit. */
-	on_shmem_exit(dsm_backend_shutdown, 0);
-
 	dsm_init_done = true;
 }
 
@@ -718,8 +723,8 @@ dsm_attach(dsm_handle h)
 /*
  * At backend shutdown time, detach any segments that are still attached.
  */
-static void
-dsm_backend_shutdown(int code, Datum arg)
+void
+dsm_backend_shutdown(void)
 {
 	while (!dlist_is_empty(&dsm_segment_list))
 	{
@@ -775,6 +780,27 @@ void
 dsm_detach(dsm_segment *seg)
 {
 	/*
+	 * Invoke registered callbacks.  Just in case one of those callbacks
+	 * throws a further error that brings us back here, pop the callback
+	 * before invoking it, to avoid infinite error recursion.
+	 */
+	while (!slist_is_empty(&seg->on_detach))
+	{
+		slist_node *node;
+		dsm_segment_detach_callback *cb;
+		on_dsm_detach_callback	function;
+		Datum		arg;
+
+		node = slist_pop_head_node(&seg->on_detach);
+		cb = slist_container(dsm_segment_detach_callback, node, node);
+		function = cb->function;
+		arg = cb->arg;
+		pfree(cb);
+
+		function(seg, arg);
+	}
+
+	/*
 	 * Try to remove the mapping, if one exists.  Normally, there will be,
 	 * but maybe not, if we failed partway through a create or attach
 	 * operation.  We remove the mapping before decrementing the reference
@@ -916,6 +942,44 @@ dsm_segment_handle(dsm_segment *seg)
 }
 
 /*
+ * Register an on-detach callback for a dynamic shared memory segment.
+ */
+void
+on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function, Datum arg)
+{
+	dsm_segment_detach_callback *cb;
+
+	cb = MemoryContextAlloc(TopMemoryContext,
+							sizeof(dsm_segment_detach_callback));
+	cb->function = function;
+	cb->arg = arg;
+	slist_push_head(&seg->on_detach, &cb->node);
+}
+
+/*
+ * Unregister an on-detach callback for a dynamic shared memory segment.
+ */
+void
+cancel_on_dsm_detach(dsm_segment *seg, on_dsm_detach_callback function,
+					 Datum arg)
+{
+	slist_mutable_iter	iter;
+
+	slist_foreach_modify(iter, &seg->on_detach)
+	{
+		dsm_segment_detach_callback *cb;
+
+		cb = slist_container(dsm_segment_detach_callback, node, iter.cur);
+		if (cb->function == function && cb->arg == arg)
+		{
+			slist_delete_current(&iter);
+			pfree(cb);
+			break;
+		}
+	}
+}
+
+/*
  * Create a segment descriptor.
  */
 static dsm_segment *
@@ -937,6 +1001,8 @@ dsm_create_descriptor(void)
 	seg->resowner = CurrentResourceOwner;
 	ResourceOwnerRememberDSM(CurrentResourceOwner, seg);
 
+	slist_init(&seg->on_detach);
+
 	return seg;
 }
 
diff --git a/src/backend/storage/ipc/ipc.c b/src/backend/storage/ipc/ipc.c
index c339e9c..9297292 100644
--- a/src/backend/storage/ipc/ipc.c
+++ b/src/backend/storage/ipc/ipc.c
@@ -27,6 +27,7 @@
 #ifdef PROFILE_PID_DIR
 #include "postmaster/autovacuum.h"
 #endif
+#include "storage/dsm.h"
 #include "storage/ipc.h"
 #include "tcop/tcopprot.h"
 
@@ -64,14 +65,19 @@ static void proc_exit_prepare(int code);
 
 #define MAX_ON_EXITS 20
 
-static struct ONEXIT
+struct ONEXIT
 {
 	pg_on_exit_callback function;
 	Datum		arg;
-}	on_proc_exit_list[MAX_ON_EXITS], on_shmem_exit_list[MAX_ON_EXITS];
+};
+
+static struct ONEXIT on_proc_exit_list[MAX_ON_EXITS];
+static struct ONEXIT on_shmem_exit_list[MAX_ON_EXITS];
+static struct ONEXIT before_shmem_exit_list[MAX_ON_EXITS];
 
 static int	on_proc_exit_index,
-			on_shmem_exit_index;
+			on_shmem_exit_index,
+			before_shmem_exit_index;
 
 
 /* ----------------------------------------------------------------
@@ -202,25 +208,60 @@ proc_exit_prepare(int code)
 /* ------------------
  * Run all of the on_shmem_exit routines --- but don't actually exit.
  * This is used by the postmaster to re-initialize shared memory and
- * semaphores after a backend dies horribly.
+ * semaphores after a backend dies horribly.  As with proc_exit(), we
+ * remove each callback from the list before calling it, to avoid
+ * infinite loop in case of error.
  * ------------------
  */
 void
 shmem_exit(int code)
 {
-	elog(DEBUG3, "shmem_exit(%d): %d callbacks to make",
-		 code, on_shmem_exit_index);
+	/*
+	 * Call before_shmem_exit callbacks.
+	 *
+	 * These should be things that need most of the system to still be
+	 * up and working, such as cleanup of temp relations, which requires
+	 * catalog access; or things that need to be completed because later
+	 * cleanup steps depend on them, such as releasing lwlocks.
+	 */
+	elog(DEBUG3, "shmem_exit(%d): %d before_shmem_exit callbacks to make",
+		 code, before_shmem_exit_index);
+	while (--before_shmem_exit_index >= 0)
+		(*before_shmem_exit_list[before_shmem_exit_index].function) (code,
+					before_shmem_exit_list[before_shmem_exit_index].arg);
+	before_shmem_exit_index = 0;
 
 	/*
-	 * call all the registered callbacks.
+	 * Call dynamic shared memory callbacks.
+	 *
+	 * These serve the same purpose as late callbacks, but for dynamic shared
+	 * memory segments rather than the main shared memory segment.
+	 * dsm_backend_shutdown() has the same kind of progressive logic we use
+	 * for the main shared memory segment; namely, it unregisters each
+	 * callback before invoking it, so that we don't get stuck in an infinite
+	 * loop if one of those callbacks itself throws an ERROR or FATAL.
+	 *
+	 * Note that explicitly calling this function here is quite different
+	 * from registering it as an on_shmem_exit callback for precisely this
+	 * reason: if one dynamic shared memory callback errors out, the remaining
+	 * callbacks will still be invoked.  Thus, hard-coding this call puts it
+	 * equal footing with callbacks for the main shared memory segment.
+	 */
+	dsm_backend_shutdown();
+
+	/*
+	 * Call on_shmem_exit callbacks.
 	 *
-	 * As with proc_exit(), we remove each callback from the list before
-	 * calling it, to avoid infinite loop in case of error.
+	 * These are generally releasing low-level shared memory resources.  In
+	 * some cases, this is a backstop against the possibility that the early
+	 * callbacks might themselves fail, leading to re-entry to this routine;
+	 * in other cases, it's cleanup that only happens at process exit.
 	 */
+	elog(DEBUG3, "shmem_exit(%d): %d on_shmem_exit callbacks to make",
+		 code, on_shmem_exit_index);
 	while (--on_shmem_exit_index >= 0)
 		(*on_shmem_exit_list[on_shmem_exit_index].function) (code,
-								on_shmem_exit_list[on_shmem_exit_index].arg);
-
+					on_shmem_exit_list[on_shmem_exit_index].arg);
 	on_shmem_exit_index = 0;
 }
 
@@ -270,10 +311,39 @@ on_proc_exit(pg_on_exit_callback function, Datum arg)
 }
 
 /* ----------------------------------------------------------------
+ *		before_shmem_exit
+ *
+ *		Register early callback to perform user-level cleanup,
+ *		e.g. transaction abort, before we begin shutting down
+ *		low-level subsystems.
+ * ----------------------------------------------------------------
+ */
+void
+before_shmem_exit(pg_on_exit_callback function, Datum arg)
+{
+	if (before_shmem_exit_index >= MAX_ON_EXITS)
+		ereport(FATAL,
+				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+				 errmsg_internal("out of before_shmem_exit slots")));
+
+	before_shmem_exit_list[before_shmem_exit_index].function = function;
+	before_shmem_exit_list[before_shmem_exit_index].arg = arg;
+
+	++before_shmem_exit_index;
+
+	if (!atexit_callback_setup)
+	{
+		atexit(atexit_callback);
+		atexit_callback_setup = true;
+	}
+}
+
+/* ----------------------------------------------------------------
  *		on_shmem_exit
  *
- *		this function adds a callback function to the list of
- *		functions invoked by shmem_exit().	-cim 2/6/90
+ *		Register ordinary callback to perform low-level shutdown
+ *		(e.g. releasing our PGPROC); run after before_shmem_exit
+ *		callbacks and before on_proc_exit callbacks.
  * ----------------------------------------------------------------
  */
 void
@@ -297,21 +367,22 @@ on_shmem_exit(pg_on_exit_callback function, Datum arg)
 }
 
 /* ----------------------------------------------------------------
- *		cancel_shmem_exit
+ *		cancel_before_shmem_exit
  *
- *		this function removes an entry, if present, from the list of
- *		functions to be invoked by shmem_exit().  For simplicity,
- *		only the latest entry can be removed.  (We could work harder
- *		but there is no need for current uses.)
+ *		this function removes a previously-registed before_shmem_exit
+ *		callback.  For simplicity, only the latest entry can be
+ *		removed.  (We could work harder but there is no need for
+ *		current uses.)
  * ----------------------------------------------------------------
  */
 void
-cancel_shmem_exit(pg_on_exit_callback function, Datum arg)
+cancel_before_shmem_exit(pg_on_exit_callback function, Datum arg)
 {
-	if (on_shmem_exit_index > 0 &&
-		on_shmem_exit_list[on_shmem_exit_index - 1].function == function &&
-		on_shmem_exit_list[on_shmem_exit_index - 1].arg == arg)
-		--on_shmem_exit_index;
+	if (before_shmem_exit_index > 0 &&
+		before_shmem_exit_list[before_shmem_exit_index - 1].function
+			== function &&
+		before_shmem_exit_list[before_shmem_exit_index - 1].arg == arg)
+		--before_shmem_exit_index;
 }
 
 /* ----------------------------------------------------------------
@@ -326,6 +397,7 @@ cancel_shmem_exit(pg_on_exit_callback function, Datum arg)
 void
 on_exit_reset(void)
 {
+	before_shmem_exit_index = 0;
 	on_shmem_exit_index = 0;
 	on_proc_exit_index = 0;
 }
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 2c7f0f1..5caa488 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -587,15 +587,14 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
 	RelationCacheInitializePhase2();
 
 	/*
-	 * Set up process-exit callback to do pre-shutdown cleanup.  This has to
-	 * be after we've initialized all the low-level modules like the buffer
-	 * manager, because during shutdown this has to run before the low-level
-	 * modules start to close down.  On the other hand, we want it in place
-	 * before we begin our first transaction --- if we fail during the
-	 * initialization transaction, as is entirely possible, we need the
-	 * AbortTransaction call to clean up.
+	 * Set up process-exit callback to do pre-shutdown cleanup.  This is the
+	 * first before_shmem_exit callback we register; thus, this will be the
+	 * last thing we do before low-level modules like the buffer manager begin
+	 * to close down.  We need to have this in place before we begin our first
+	 * transaction --- if we fail during the initialization transaction, as is
+	 * entirely possible, we need the AbortTransaction call to clean up.
 	 */
-	on_shmem_exit(ShutdownPostgres, 0);
+	before_shmem_exit(ShutdownPostgres, 0);
 
 	/* The autovacuum launcher is done here */
 	if (IsAutoVacuumLauncherProcess())
diff --git a/src/include/storage/dsm.h b/src/include/storage/dsm.h
index d906eba..545ac81 100644
--- a/src/include/storage/dsm.h
+++ b/src/include/storage/dsm.h
@@ -17,8 +17,9 @@
 
 typedef struct dsm_segment dsm_segment;
 
-/* Initialization function. */
+/* Startup and shutdown functions. */
 extern void dsm_postmaster_startup(void);
+extern void dsm_backend_shutdown(void);
 
 /* Functions that create, update, or remove mappings. */
 extern dsm_segment *dsm_create(Size size);
@@ -36,4 +37,11 @@ extern void *dsm_segment_address(dsm_segment *seg);
 extern Size dsm_segment_map_length(dsm_segment *seg);
 extern dsm_handle dsm_segment_handle(dsm_segment *seg);
 
+/* Cleanup hooks. */
+typedef void (*on_dsm_detach_callback) (dsm_segment *, Datum arg);
+extern void on_dsm_detach(dsm_segment *seg,
+			  on_dsm_detach_callback function, Datum arg);
+extern void cancel_on_dsm_detach(dsm_segment *seg,
+			  on_dsm_detach_callback function, Datum arg);
+
 #endif   /* DSM_H */
diff --git a/src/include/storage/ipc.h b/src/include/storage/ipc.h
index ac4ac66..46dbd7b 100644
--- a/src/include/storage/ipc.h
+++ b/src/include/storage/ipc.h
@@ -46,14 +46,14 @@ typedef void (*shmem_startup_hook_type) (void);
  */
 #define PG_ENSURE_ERROR_CLEANUP(cleanup_function, arg)	\
 	do { \
-		on_shmem_exit(cleanup_function, arg); \
+		before_shmem_exit(cleanup_function, arg); \
 		PG_TRY()
 
 #define PG_END_ENSURE_ERROR_CLEANUP(cleanup_function, arg)	\
-		cancel_shmem_exit(cleanup_function, arg); \
+		cancel_before_shmem_exit(cleanup_function, arg); \
 		PG_CATCH(); \
 		{ \
-			cancel_shmem_exit(cleanup_function, arg); \
+			cancel_before_shmem_exit(cleanup_function, arg); \
 			cleanup_function (0, arg); \
 			PG_RE_THROW(); \
 		} \
@@ -68,7 +68,8 @@ extern void proc_exit(int code) __attribute__((noreturn));
 extern void shmem_exit(int code);
 extern void on_proc_exit(pg_on_exit_callback function, Datum arg);
 extern void on_shmem_exit(pg_on_exit_callback function, Datum arg);
-extern void cancel_shmem_exit(pg_on_exit_callback function, Datum arg);
+extern void before_shmem_exit(pg_on_exit_callback function, Datum arg);
+extern void cancel_before_shmem_exit(pg_on_exit_callback function, Datum arg);
 extern void on_exit_reset(void);
 
 /* ipci.c */

commit 7deaabe19d32f71f8989877b68cda5e2776da044
Author: Robert Haas <rhaas@postgresql.org>
Date:   Tue Aug 27 14:23:30 2013 -0400

    Single-reader, single-writer, lightweight shared message queue.
    
    This code provides infrastructure for user backends to communicate
    relatively easily with background workers.  The message queue is
    structured as a ring buffer and allows messages of arbitary length
    to be sent and received.

diff --git a/src/backend/storage/ipc/Makefile b/src/backend/storage/ipc/Makefile
index df0a49e..850347c 100644
--- a/src/backend/storage/ipc/Makefile
+++ b/src/backend/storage/ipc/Makefile
@@ -16,6 +16,6 @@ endif
 endif
 
 OBJS = dsm_impl.o dsm.o ipc.o ipci.o pmsignal.o procarray.o procsignal.o \
-	shmem.o shmqueue.o shm_toc.o sinval.o sinvaladt.o standby.o
+	shmem.o shmqueue.o shm_mq.o shm_toc.o sinval.o sinvaladt.o standby.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
new file mode 100644
index 0000000..26d7b7c
--- /dev/null
+++ b/src/backend/storage/ipc/shm_mq.c
@@ -0,0 +1,936 @@
+/*-------------------------------------------------------------------------
+ *
+ * shm_mq.c
+ *	  single-reader, single-writer shared memory message queue
+ *
+ * Both the sender and the receiver must have a PGPROC; their respective
+ * process latches are used for synchronization.  Only the sender may send,
+ * and only the receiver may receive.  This is intended to allow a user
+ * backend to communicate with worker backends that it has registered.
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/storage/shm_mq.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/procsignal.h"
+#include "storage/shm_mq.h"
+#include "storage/spin.h"
+
+/*
+ * Some notes on synchronization:
+ *
+ * mq_receiver and mq_bytes_read can only be changed by the receiver; and
+ * mq_sender and mq_bytes_written can only be changed by the sender.  However,
+ * because most of these fields are 8 bytes and we don't assume that 8 byte
+ * reads and writes are atomic, the spinlock must be taken whenever the field
+ * is updated, and whenever it is read by a process other than the one allowed
+ * to modify it. But the process that is allowed to modify it is also allowed
+ * to read it without the lock.  On architectures where 8-byte writes are
+ * atomic, we could replace these spinlocks with memory barriers, but
+ * testing found no performance benefit, so it seems best to keep things
+ * simple for now.
+ *
+ * mq_detached can be set by either the sender or the receiver, so the mutex
+ * must be held to read or write it.  Memory barriers could be used here as
+ * well, if needed.
+ *
+ * mq_ring_size and mq_ring_offset never change after initialization, and
+ * can therefore be read without the lock.
+ *
+ * Importantly, mq_ring can be safely read and written without a lock.  Were
+ * this not the case, we'd have to hold the spinlock for much longer
+ * intervals, and performance might suffer.  Fortunately, that's not
+ * necessary.  At any given time, the difference between mq_bytes_read and
+ * mq_bytes_written defines the number of bytes within mq_ring that contain
+ * unread data, and mq_bytes_read defines the position where those bytes
+ * begin.  The sender can increase the number of unread bytes at any time,
+ * but only the receiver can give license to overwrite those bytes, by
+ * incrementing mq_bytes_read.  Therefore, it's safe for the receiver to read
+ * the unread bytes it knows to be present without the lock.  Conversely,
+ * the sender can write to the unused portion of the ring buffer without
+ * the lock, because nobody else can be reading or writing those bytes.  The
+ * receiver could be making more bytes unused by incrementing mq_bytes_read,
+ * but that's OK.  Note that it would be unsafe for the receiver to read any
+ * data it's already marked as read, or to write any data; and it would be
+ * unsafe for the sender to reread any data after incrementing
+ * mq_bytes_written, but fortunately there's no need for any of that.
+ */
+struct shm_mq
+{
+	slock_t		mq_mutex;
+	PGPROC	   *mq_receiver;
+	PGPROC	   *mq_sender;
+	uint64		mq_bytes_read;
+	uint64		mq_bytes_written;
+	uint64		mq_ring_size;
+	bool		mq_detached;
+	uint8		mq_ring_offset;
+	char		mq_ring[FLEXIBLE_ARRAY_MEMBER];
+};
+
+/*
+ * mqh_queue is a pointer to the queue we've attached, and mqh_segment is
+ * a pointer to the dynamic shared memory segment that contains it.
+ *
+ * If this queue is intended to connect the current process with a background
+ * worker that started it, the user can pass a pointer to the worker handle
+ * to shm_mq_attach(), and we'll store it in mqh_handle.  The point of this
+ * is to allow us to begin sending to or receiving from that queue before the
+ * process we'll be communicating with has even been started.  If it fails
+ * to start, the handle will allow us to notice that and fail cleanly, rather
+ * than waiting forever; see shm_mq_wait_internal.  This is mostly useful in
+ * simple cases - e.g. where there are just 2 processes communicating; in
+ * more complex scenarios, every process may not have a BackgroundWorkerHandle
+ * available, or may need to watch for the failure of more than one other
+ * process at a time.
+ *
+ * When a message exists as a contiguous chunk of bytes in the queue - that is,
+ * it is smaller than the size of the ring buffer and does not wrap around
+ * the end - we return the message to the caller as a pointer into the buffer.
+ * For messages that are larger or happen to wrap, we reassemble the message
+ * locally by copying the chunks into a backend-local buffer.  mqh_buffer is
+ * the buffer, and mqh_buflen is the number of bytes allocated for it.
+ *
+ * mqh_partial_message_bytes, mqh_expected_bytes, and mqh_did_length_word
+ * are used to track the state of non-blocking operations.  When the caller
+ * attempts a non-blocking operation that returns SHM_MQ_WOULD_BLOCK, they
+ * are expected to retry the call at a later time with the same argument;
+ * we need to retain enough state to pick up where we left off.
+ * mqh_did_length_word tracks whether we read or wrote the length word,
+ * mqh_partial_message_bytes tracks the number of payload bytes read or
+ * written, and mqh_expected_bytes - which is used only for reads - tracks
+ * the expected total size of the payload.
+ *
+ * mqh_counterparty_attached tracks whether we know the counterparty to have
+ * attached to the queue at some previous point.  This lets us avoid some
+ * mutex acquisitions.
+ *
+ * mqh_context is the memory context in effect at the time we attached to
+ * the shm_mq.  The shm_mq_handle itself is allocated in this context, and
+ * we make sure any other allocations we do happen in this context as well,
+ * to avoid nasty surprises.
+ */
+struct shm_mq_handle
+{
+	shm_mq	   *mqh_queue;
+	dsm_segment *mqh_segment;
+	BackgroundWorkerHandle *mqh_handle;
+	char	   *mqh_buffer;
+	uint64		mqh_buflen;
+	uint64		mqh_consume_pending;
+	uint64		mqh_partial_message_bytes;
+	uint64		mqh_expected_bytes;
+	bool		mqh_did_length_word;
+	bool		mqh_counterparty_attached;
+	MemoryContext mqh_context;
+};
+
+static shm_mq_result shm_mq_send_bytes(shm_mq_handle *mq, uint64 nbytes,
+				  void *data, bool nowait, uint64 *bytes_written);
+static shm_mq_result shm_mq_receive_bytes(shm_mq *mq, uint64 bytes_needed,
+					 bool nowait, uint64 *nbytesp, void **datap);
+static bool shm_mq_wait_internal(volatile shm_mq *mq, PGPROC * volatile *ptr,
+					 BackgroundWorkerHandle *handle);
+static uint64 shm_mq_get_bytes_read(volatile shm_mq *mq, bool *detached);
+static void shm_mq_inc_bytes_read(volatile shm_mq *mq, uint64 n);
+static uint64 shm_mq_get_bytes_written(volatile shm_mq *mq, bool *detached);
+static void shm_mq_inc_bytes_written(volatile shm_mq *mq, uint64 n);
+static shm_mq_result shm_mq_notify_receiver(volatile shm_mq *mq);
+static void shm_mq_detach_callback(dsm_segment *seg, Datum arg);
+
+/* Minimum queue size is enough for header and at least one chunk of data. */
+Size shm_mq_minimum_size =
+	MAXALIGN(offsetof(shm_mq, mq_ring)) + MAXIMUM_ALIGNOF;
+
+#define MQH_BUFSIZE				8192
+
+/*
+ * Initialize a new shared message queue.
+ */
+shm_mq *
+shm_mq_create(void *address, Size size)
+{
+	shm_mq	   *mq = address;
+	uint64		data_offset = MAXALIGN(offsetof(shm_mq, mq_ring));
+
+	/* If the size isn't MAXALIGN'd, just discard the odd bytes. */
+	size = MAXALIGN_DOWN(size);
+
+	/* Queue size must be large enough to hold some data. */
+	Assert(size > data_offset);
+
+	/* Initialize queue header. */
+	mq->mq_receiver = NULL;
+	mq->mq_sender = NULL;
+	mq->mq_bytes_read = 0;
+	mq->mq_bytes_written = 0;
+	mq->mq_ring_size = size - data_offset;
+	mq->mq_detached = false;
+	mq->mq_ring_offset = data_offset - offsetof(shm_mq, mq_ring);
+
+	return mq;
+}
+
+/*
+ * Set the identity of the process that will receive from a shared message
+ * queue.
+ */
+void
+shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *sender;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	Assert(vmq->mq_receiver == NULL);
+	vmq->mq_receiver = proc;
+	sender = vmq->mq_sender;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (sender != NULL)
+		SetLatch(&sender->procLatch);
+}
+
+/*
+ * Set the identity of the process that will send to a shared message queue.
+ */
+void
+shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *receiver;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	Assert(vmq->mq_sender == NULL);
+	vmq->mq_sender = proc;
+	receiver = vmq->mq_receiver;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (receiver != NULL)
+		SetLatch(&receiver->procLatch);
+}
+
+/*
+ * Get the configured receiver.
+ */
+PGPROC *
+shm_mq_get_receiver(shm_mq *mq)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *receiver;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	receiver = vmq->mq_receiver;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return receiver;
+}
+
+/*
+ * Get the configured sender.
+ */
+PGPROC *
+shm_mq_get_sender(shm_mq *mq)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *sender;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	sender = vmq->mq_sender;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return sender;
+}
+
+/*
+ * Attach to a shared message queue so we can send or receive messages.
+ *
+ * The memory context in effect at the time this function is called should
+ * be one which will last for at least as long as the message queue itself.
+ *
+ * If seg != NULL, the queue will be automatically detached when that dynamic
+ * shared memory segment is detached.
+ *
+ * If handle != NULL, the queue can be read or written even before the
+ * other process has attached.  We'll wait for it to do so if needed.  The
+ * handle must be for a background worker initialized with bgw_notify_pid
+ * equal to our PID.
+ *
+ * shm_mq_detach() should be called when done.  This will free the
+ * shm_mq_handle and mark the queue itself as detached, so that our
+ * counterpart won't get stuck waiting for us to fill or drain the queue
+ * after we've already lost interest.
+ */
+shm_mq_handle *
+shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
+{
+	shm_mq_handle	   *mqh = palloc(sizeof(shm_mq_handle));
+
+	Assert(mq->mq_receiver == MyProc || mq->mq_sender == MyProc);
+	mqh->mqh_queue = mq;
+	mqh->mqh_segment = seg;
+	mqh->mqh_buffer = NULL;
+	mqh->mqh_handle = handle;
+	mqh->mqh_buflen = 0;
+	mqh->mqh_consume_pending = 0;
+	mqh->mqh_context = CurrentMemoryContext;
+	mqh->mqh_partial_message_bytes = 0;
+	mqh->mqh_did_length_word = false;
+	mqh->mqh_counterparty_attached = false;
+
+	if (seg != NULL)
+		on_dsm_detach(seg, shm_mq_detach_callback, PointerGetDatum(mq));
+
+	return mqh;
+}
+
+/*
+ * Write a message into a shared message queue.
+ *
+ * When nowait = false, we'll wait on our process latch when the ring buffer
+ * fills up, and then continue writing once the receiver has drained some data.
+ * The process latch is reset after each wait.
+ *
+ * When nowait = true, we do not manipulate the state of the process latch;
+ * instead, if the buffer becomes full, we return SHM_MQ_WOULD_BLOCK.  In
+ * this case, the caller should call this function again, with the same
+ * arguments, each time the process latch is set.
+ */
+shm_mq_result
+shm_mq_send(shm_mq_handle *mqh, uint64 nbytes, void *data, bool nowait)
+{
+	shm_mq_result	res;
+	shm_mq		   *mq = mqh->mqh_queue;
+	uint64			bytes_written;
+
+	Assert(mq->mq_sender == MyProc);
+
+	/* Write the message length into the buffer. */
+	if (!mqh->mqh_did_length_word)
+	{
+		res = shm_mq_send_bytes(mqh, sizeof(uint64), &nbytes, nowait,
+								&bytes_written);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+
+		/*
+		 * We're sure to have sent the length in full, since we always
+		 * write a MAXALIGN'd chunk.
+		 */
+		Assert(bytes_written == MAXALIGN64(sizeof(uint64)));
+		mqh->mqh_did_length_word = true;
+	}
+
+	/* Write the actual data bytes into the buffer. */
+	Assert(mqh->mqh_partial_message_bytes <= nbytes);
+	res = shm_mq_send_bytes(mqh, nbytes - mqh->mqh_partial_message_bytes,
+							((char *) data) + mqh->mqh_partial_message_bytes,
+							nowait, &bytes_written);
+	if (res == SHM_MQ_WOULD_BLOCK)
+		mqh->mqh_partial_message_bytes += bytes_written;
+	else
+	{
+		mqh->mqh_partial_message_bytes = 0;
+		mqh->mqh_did_length_word = false;
+	}
+	if (res != SHM_MQ_SUCCESS)
+		return res;
+
+	/* Notify receiver of the newly-written data, and return. */
+	return shm_mq_notify_receiver(mq);
+}
+
+/*
+ * Receive a message from a shared message queue.
+ *
+ * We set *nbytes to the message length and *data to point to the message
+ * payload.  If the entire message exists in the queue as a single,
+ * contiguous chunk, *data will point directly into shared memory; otherwise,
+ * it will point to a temporary buffer.  This mostly avoids data copying in
+ * the hoped-for case where messages are short compared to the buffer size,
+ * while still allowing longer messages.  In either case, the return value
+ * remains valid until the next receive operation is perfomed on the queue.
+ *
+ * When nowait = false, we'll wait on our process latch when the ring buffer
+ * is empty and we have not yet received a full message.  The sender will
+ * set our process latch after more data has been written, and we'll resume
+ * processing.  Each call will therefore return a complete message
+ * (unless the sender detaches the queue).
+ *
+ * When nowait = true, we do not manipulate the state of the process latch;
+ * instead, whenever the buffer is empty and we need to read from it, we
+ * return SHM_MQ_WOULD_BLOCK.  In this case, the caller should call this
+ * function again after the process latch has been set.
+ */
+shm_mq_result
+shm_mq_receive(shm_mq_handle *mqh, uint64 *nbytesp, void **datap, bool nowait)
+{
+	shm_mq		   *mq = mqh->mqh_queue;
+	shm_mq_result	res;
+	uint64			rb = 0;
+	uint64			nbytes;
+	uint64			needed;
+	void		   *rawdata;
+
+	Assert(mq->mq_receiver == MyProc);
+
+	/* We can't receive data until the sender has attached. */
+	if (!mqh->mqh_counterparty_attached)
+	{
+		if (nowait)
+		{
+			if (shm_mq_get_sender(mq) == NULL)
+				return SHM_MQ_WOULD_BLOCK;
+		}
+		else if (!shm_mq_wait_internal(mq, &mq->mq_sender, mqh->mqh_handle))
+		{
+			mq->mq_detached = true;
+			return SHM_MQ_DETACHED;
+		}
+		mqh->mqh_counterparty_attached = true;
+	}
+
+	/* Consume any zero-copy data from previous receive operation. */
+	if (mqh->mqh_consume_pending > 0)
+	{
+		shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
+		mqh->mqh_consume_pending = 0;
+	}
+
+	/* Determine the message length. */
+	if (mqh->mqh_did_length_word)
+	{
+		/* We've partially received a message; recall expected length. */
+		nbytes = mqh->mqh_expected_bytes;
+	}
+	else
+	{
+		/* Try to receive the message length word. */
+		res = shm_mq_receive_bytes(mq, sizeof(uint64), nowait, &rb, &rawdata);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+		Assert(rb >= sizeof(uint64));
+		memcpy(&nbytes, rawdata, sizeof(uint64));
+		mqh->mqh_expected_bytes = nbytes;
+
+		/* If we've already got the whole message, we're done. */
+		needed = MAXALIGN64(sizeof(uint64)) + MAXALIGN64(nbytes);
+		if (rb >= needed)
+		{
+			/*
+			 * Technically, we could consume the message length information at
+			 * this point, but the extra write to shared memory wouldn't be
+			 * free and in most cases we would reap no benefit.
+			 */
+			mqh->mqh_consume_pending = needed;
+			*nbytesp = nbytes;
+			*datap = ((char *) rawdata) + MAXALIGN64(sizeof(uint64));
+			return SHM_MQ_SUCCESS;
+		}
+
+		/* Consume the length word. */
+		shm_mq_inc_bytes_read(mq, MAXALIGN64(sizeof(uint64)));
+		mqh->mqh_did_length_word = true;
+		rb -= MAXALIGN64(sizeof(uint64));
+	}
+
+	if (mqh->mqh_partial_message_bytes == 0)
+	{
+		/*
+		 * Try to obtain the whole message in a single chunk.  If this works,
+		 * we need not copy the data and can return a pointer directly into
+		 * shared memory.
+		 */
+		res = shm_mq_receive_bytes(mq, nbytes, nowait, &rb, &rawdata);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+		if (rb >= nbytes)
+		{
+			mqh->mqh_did_length_word = false;
+			mqh->mqh_consume_pending = MAXALIGN64(nbytes);
+			*nbytesp = nbytes;
+			*datap = rawdata;
+			return SHM_MQ_SUCCESS;
+		}
+
+		/*
+		 * The message has wrapped the buffer.  We'll need to copy it in order
+		 * to return it to the client in one chunk.  First, make sure we have a
+		 * large enough buffer available.
+		 */
+		if (mqh->mqh_buflen < nbytes)
+		{
+			uint64		newbuflen = Max(mqh->mqh_buflen, MQH_BUFSIZE);
+
+			while (newbuflen < nbytes)
+				newbuflen *= 2;
+
+			if (mqh->mqh_buffer != NULL)
+			{
+				pfree(mqh->mqh_buffer);
+				mqh->mqh_buffer = NULL;
+				mqh->mqh_buflen = 0;
+			}
+			mqh->mqh_buffer = MemoryContextAlloc(mqh->mqh_context, newbuflen);
+			mqh->mqh_buflen = newbuflen;
+		}
+	}
+
+	/* Loop until we've copied the entire message. */
+	for (;;)
+	{
+		uint64	still_needed;
+
+		/* Copy as much as we can. */
+		Assert(mqh->mqh_partial_message_bytes + rb <= nbytes);
+		memcpy(&mqh->mqh_buffer[mqh->mqh_partial_message_bytes], rawdata, rb);
+		mqh->mqh_partial_message_bytes += rb;
+
+		/*
+		 * Update count of bytes read, with alignment padding.  Note
+		 * that this will never actually insert any padding except at the
+		 * end of a message, because the buffer size is a multiple of
+		 * MAXIMUM_ALIGNOF, and each read and write is as well.
+		 */
+		Assert(mqh->mqh_partial_message_bytes == nbytes ||
+				rb == MAXALIGN64(rb));
+		shm_mq_inc_bytes_read(mq, MAXALIGN64(rb));
+
+		/* If we got all the data, exit the loop. */
+		if (mqh->mqh_partial_message_bytes >= nbytes)
+			break;
+
+		/* Wait for some more data. */
+		still_needed = nbytes - mqh->mqh_partial_message_bytes;
+		res = shm_mq_receive_bytes(mq, still_needed, nowait, &rb, &rawdata);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+		if (rb > still_needed)
+			rb = still_needed;
+	}
+
+	/* Return the complete message, and reset for next message. */
+	*nbytesp = nbytes;
+	*datap = mqh->mqh_buffer;
+	mqh->mqh_did_length_word = false;
+	mqh->mqh_partial_message_bytes = 0;
+	return SHM_MQ_SUCCESS;
+}
+
+/*
+ * Wait for the other process that's supposed to use this queue to attach
+ * to it.
+ *
+ * The return value is SHM_MQ_DETACHED if the worker has already detached or
+ * if it dies; it is SHM_MQ_SUCCESS if we detect that the worker has attached.
+ * Note that we will only be able to detect that the worker has died before
+ * attaching if a background worker handle was passed to shm_mq_attach().
+ */
+shm_mq_result
+shm_mq_wait_for_attach(shm_mq_handle *mqh)
+{
+	shm_mq	   *mq = mqh->mqh_queue;
+	PGPROC	   **victim;
+
+	if (shm_mq_get_receiver(mq) == MyProc)
+		victim = &mq->mq_sender;
+	else
+	{
+		Assert(shm_mq_get_sender(mq) == MyProc);
+		victim = &mq->mq_receiver;
+	}
+
+	if (shm_mq_wait_internal(mq, victim, mqh->mqh_handle))
+		return SHM_MQ_SUCCESS;
+	else
+		return SHM_MQ_DETACHED;
+}
+
+/*
+ * Detach a shared message queue.
+ *
+ * The purpose of this function is to make sure that the process
+ * with which we're communicating doesn't block forever waiting for us to
+ * fill or drain the queue once we've lost interest.  Whem the sender
+ * detaches, the receiver can read any messages remaining in the queue;
+ * further reads will return SHM_MQ_DETACHED.  If the receiver detaches,
+ * further attempts to send messages will likewise return SHM_MQ_DETACHED.
+ */
+void
+shm_mq_detach(shm_mq *mq)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC	   *victim;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	if (vmq->mq_sender == MyProc)
+		victim = vmq->mq_receiver;
+	else
+	{
+		Assert(vmq->mq_receiver == MyProc);
+		victim = vmq->mq_sender;
+	}
+	vmq->mq_detached = true;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (victim != NULL)
+		SetLatch(&victim->procLatch);
+}
+
+/*
+ * Write bytes into a shared message queue.
+ */
+static shm_mq_result
+shm_mq_send_bytes(shm_mq_handle *mqh, uint64 nbytes, void *data, bool nowait,
+				  uint64 *bytes_written)
+{
+	shm_mq	   *mq = mqh->mqh_queue;
+	uint64		sent = 0;
+	uint64		used;
+	uint64		ringsize = mq->mq_ring_size;
+	uint64		available;
+
+	while (sent < nbytes)
+	{
+		bool	detached;
+		uint64	rb;
+
+		/* Compute number of ring buffer bytes used and available. */
+		rb = shm_mq_get_bytes_read(mq, &detached);
+		Assert(mq->mq_bytes_written >= rb);
+		used = mq->mq_bytes_written - rb;
+		Assert(used <= ringsize);
+		available = Min(ringsize - used, nbytes - sent);
+
+		/* Bail out if the queue has been detached. */
+		if (detached)
+			return SHM_MQ_DETACHED;
+
+		if (available == 0)
+		{
+			shm_mq_result	res;
+
+			/*
+			 * The queue is full, so if the receiver isn't yet known to be
+			 * attached, we must wait for that to happen.
+			 */
+			if (!mqh->mqh_counterparty_attached)
+			{
+				if (nowait)
+				{
+					if (shm_mq_get_receiver(mq) == NULL)
+						return SHM_MQ_WOULD_BLOCK;
+				}
+				else if (!shm_mq_wait_internal(mq, &mq->mq_receiver,
+											   mqh->mqh_handle))
+				{
+					mq->mq_detached = true;
+					return SHM_MQ_DETACHED;
+				}
+				mqh->mqh_counterparty_attached = true;
+			}
+
+			/* Let the receiver know that we need them to read some data. */
+			res = shm_mq_notify_receiver(mq);
+			if (res != SHM_MQ_SUCCESS)
+			{
+				*bytes_written = res;
+				return res;
+			}
+
+			/* Skip manipulation of our latch if nowait = true. */
+			if (nowait)
+			{
+				*bytes_written = sent;
+				return SHM_MQ_WOULD_BLOCK;
+			}
+
+			/*
+			 * Wait for our latch to be set.  It might already be set for
+			 * some unrelated reason, but that'll just result in one extra
+			 * trip through the loop.  It's worth it to avoid resetting the
+			 * latch at top of loop, because setting an already-set latch is
+			 * much cheaper than setting one that has been reset.
+			 */
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+			/* An interrupt may have occurred while we were waiting. */
+			CHECK_FOR_INTERRUPTS();
+
+			/* Reset the latch so we don't spin. */
+			ResetLatch(&MyProc->procLatch);
+		}
+		else
+		{
+			uint64	offset = mq->mq_bytes_written % ringsize;
+			uint64	sendnow = Min(available, ringsize - offset);
+
+			/* Write as much data as we can via a single memcpy(). */
+			memcpy(&mq->mq_ring[mq->mq_ring_offset + offset],
+				   (char *) data + sent, sendnow);
+			sent += sendnow;
+
+			/*
+			 * Update count of bytes written, with alignment padding.  Note
+			 * that this will never actually insert any padding except at the
+			 * end of a run of bytes, because the buffer size is a multiple of
+			 * MAXIMUM_ALIGNOF, and each read is as well.
+			 */
+			Assert(sent == nbytes || sendnow == MAXALIGN64(sendnow));
+			shm_mq_inc_bytes_written(mq, MAXALIGN64(sendnow));
+
+			/*
+			 * For efficiency, we don't set the reader's latch here.  We'll
+			 * do that only when the buffer fills up or after writing an
+			 * entire message.
+			 */
+		}
+	}
+
+	*bytes_written = sent;
+	return SHM_MQ_SUCCESS;
+}
+
+/*
+ * Wait until at least *nbytesp bytes are available to be read from the
+ * shared message queue, or until the buffer wraps around.  On return,
+ * *datap is set to the location at which data bytes can be read.  The
+ * return value is the number of bytes available to be read starting at
+ * that offset; if the message has wrapped the buffer, it may be less than
+ * bytes_needed.
+ */
+static shm_mq_result
+shm_mq_receive_bytes(shm_mq *mq, uint64 bytes_needed, bool nowait,
+					 uint64 *nbytesp, void **datap)
+{
+	uint64		used;
+	uint64		ringsize = mq->mq_ring_size;
+	uint64		written;
+
+	for (;;)
+	{
+		uint64		offset;
+		bool		detached;
+
+		/* Get bytes written, so we can compute what's available to read. */
+		written = shm_mq_get_bytes_written(mq, &detached);
+		used = written - mq->mq_bytes_read;
+		Assert(used <= ringsize);
+		offset = mq->mq_bytes_read % ringsize;
+
+		/* If we have enough data or buffer has wrapped, we're done. */
+		if (used >= bytes_needed || offset + used >= ringsize)
+		{
+			*nbytesp = Min(used, ringsize - offset);
+			*datap = &mq->mq_ring[mq->mq_ring_offset + offset];
+			return SHM_MQ_SUCCESS;
+		}
+
+		/*
+		 * Fall out before waiting if the queue has been detached.
+		 *
+		 * Note that we don't check for this until *after* considering
+		 * whether the data already available is enough, since the
+		 * receiver can finish receiving a message stored in the buffer
+		 * even after the sender has detached.
+		 */
+		if (detached)
+			return SHM_MQ_DETACHED;
+
+		/* Skip manipulation of our latch if nowait = true. */
+		if (nowait)
+			return SHM_MQ_WOULD_BLOCK;
+
+		/*
+		 * Wait for our latch to be set.  It might already be set for
+		 * some unrelated reason, but that'll just result in one extra
+		 * trip through the loop.  It's worth it to avoid resetting the
+		 * latch at top of loop, because setting an already-set latch is
+		 * much cheaper than setting one that has been reset.
+		 */
+		WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+		/* An interrupt may have occurred while we were waiting. */
+		CHECK_FOR_INTERRUPTS();
+
+		/* Reset the latch so we don't spin. */
+		ResetLatch(&MyProc->procLatch);
+	}
+}
+
+/*
+ * This is used when a process is waiting for its counterpart to attach to the
+ * queue.  We exit when the other process attaches as expected, or, if
+ * handle != NULL, when the referenced background process or the postmaster
+ * dies.  Note that if handle == NULL, and the process fails to attach, we'll
+ * potentially get stuck here forever waiting for a process that may never
+ * start.  We do check for interrupts, though.
+ *
+ * ptr is a pointer to the memory address that we're expecting to become
+ * non-NULL when our counterpart attaches to the queue.
+ */
+static bool
+shm_mq_wait_internal(volatile shm_mq *mq, PGPROC * volatile *ptr,
+					 BackgroundWorkerHandle *handle)
+{
+	bool	save_set_latch_on_sigusr1;
+	bool	result = false;
+
+	save_set_latch_on_sigusr1 = set_latch_on_sigusr1;
+	if (handle != NULL)
+		set_latch_on_sigusr1 = true;
+
+	PG_TRY();
+	{
+		for (;;)
+		{
+			BgwHandleStatus	status;
+			pid_t	pid;
+			bool	detached;
+
+			/* Acquire the lock just long enough to check the pointer. */
+			SpinLockAcquire(&mq->mq_mutex);
+			detached = mq->mq_detached;
+			result = (*ptr != NULL);
+			SpinLockRelease(&mq->mq_mutex);
+
+			/* Fail if detached; else succeed if initialized. */
+			if (detached)
+			{
+				result = false;
+				break;
+			}
+			if (result)
+				break;
+
+			if (handle != NULL)
+			{
+				/* Check for unexpected worker death. */
+				status = GetBackgroundWorkerPid(handle, &pid);
+				if (status != BGWH_STARTED && status != BGWH_NOT_YET_STARTED)
+				{
+					result = false;
+					break;
+				}
+			}
+
+			/* Wait to be signalled. */
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+			/* An interrupt may have occurred while we were waiting. */
+			CHECK_FOR_INTERRUPTS();
+
+			/* Reset the latch so we don't spin. */
+			ResetLatch(&MyProc->procLatch);
+		}
+	}
+	PG_CATCH();
+	{
+		set_latch_on_sigusr1 = save_set_latch_on_sigusr1;
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
+
+	return result;
+}
+
+/*
+ * Get the number of bytes read.  The receiver need not use this to access
+ * the count of bytes read, but the sender must.
+ */
+static uint64
+shm_mq_get_bytes_read(volatile shm_mq *mq, bool *detached)
+{
+	uint64	v;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	v = mq->mq_bytes_read;
+	*detached = mq->mq_detached;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return mq->mq_bytes_read;
+}
+
+/*
+ * Increment the number of bytes read.
+ */
+static void
+shm_mq_inc_bytes_read(volatile shm_mq *mq, uint64 n)
+{
+	PGPROC	   *sender;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	mq->mq_bytes_read += n;
+	sender = mq->mq_sender;
+	SpinLockRelease(&mq->mq_mutex);
+
+	/* We shoudn't have any bytes to read without a sender. */
+	Assert(sender != NULL);
+	SetLatch(&sender->procLatch);
+}
+
+/*
+ * Get the number of bytes written.  The sender need not use this to access
+ * the count of bytes written, but the reciever must.
+ */
+static uint64
+shm_mq_get_bytes_written(volatile shm_mq *mq, bool *detached)
+{
+	uint64	v;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	v = mq->mq_bytes_written;
+	*detached = mq->mq_detached;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return mq->mq_bytes_written;
+}
+
+/*
+ * Increment the number of bytes written.
+ */
+static void
+shm_mq_inc_bytes_written(volatile shm_mq *mq, uint64 n)
+{
+	SpinLockAcquire(&mq->mq_mutex);
+	mq->mq_bytes_written += n;
+	SpinLockRelease(&mq->mq_mutex);
+}
+
+/*
+ * Set sender's latch, unless queue is detached.
+ */
+static shm_mq_result
+shm_mq_notify_receiver(volatile shm_mq *mq)
+{
+	PGPROC *receiver;
+	bool	detached;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	detached = mq->mq_detached;
+	receiver = mq->mq_receiver;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (detached)
+		return SHM_MQ_DETACHED;
+	if (receiver)
+		SetLatch(&receiver->procLatch);
+	return SHM_MQ_SUCCESS;
+}
+
+/* Shim for on_dsm_callback. */
+static void
+shm_mq_detach_callback(dsm_segment *seg, Datum arg)
+{
+	shm_mq	   *mq = (shm_mq *) DatumGetPointer(arg);
+
+	shm_mq_detach(mq);
+}
diff --git a/src/include/storage/shm_mq.h b/src/include/storage/shm_mq.h
new file mode 100644
index 0000000..749ab3b
--- /dev/null
+++ b/src/include/storage/shm_mq.h
@@ -0,0 +1,70 @@
+/*-------------------------------------------------------------------------
+ *
+ * shm_mq.h
+ *	  single-reader, single-writer shared memory message queue
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/storage/shm_mq.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SHM_MQ_H
+#define SHM_MQ_H
+
+#include "postmaster/bgworker.h"
+#include "storage/dsm.h"
+#include "storage/proc.h"
+
+/* The queue itself, in shared memory. */
+struct shm_mq;
+typedef struct shm_mq shm_mq;
+
+/* Backend-private state. */
+struct shm_mq_handle;
+typedef struct shm_mq_handle shm_mq_handle;
+
+/* Possible results of a send or receive operation. */
+typedef enum
+{
+	SHM_MQ_SUCCESS,			/* Sent or received a message. */
+	SHM_MQ_WOULD_BLOCK,		/* Not completed; retry later. */
+	SHM_MQ_DETACHED			/* Other process has detached queue. */
+} shm_mq_result;
+
+/*
+ * Primitives to create a queue and set the sender and receiver.
+ *
+ * Both the sender and the receiver must be set before any messages are read
+ * or written, but they need not be set by the same process.  Each must be
+ * set exactly once.
+ */
+extern shm_mq *shm_mq_create(void *address, Size size);
+extern void shm_mq_set_receiver(shm_mq *mq, PGPROC *);
+extern void shm_mq_set_sender(shm_mq *mq, PGPROC *);
+
+/* Accessor methods for sender and receiver. */
+extern PGPROC *shm_mq_get_receiver(shm_mq *);
+extern PGPROC *shm_mq_get_sender(shm_mq *);
+
+/* Set up backend-local queue state. */
+extern shm_mq_handle *shm_mq_attach(shm_mq *mq, dsm_segment *seg,
+			  BackgroundWorkerHandle *handle);
+
+/* Break connection. */
+extern void shm_mq_detach(shm_mq *);
+
+/* Send or receive messages. */
+extern shm_mq_result shm_mq_send(shm_mq_handle *mqh,
+			uint64 nbytes, void *data, bool nowait);
+extern shm_mq_result shm_mq_receive(shm_mq_handle *mqh,
+			   uint64 *nbytesp, void **datap, bool nowait);
+
+/* Wait for our counterparty to attach to the queue. */
+extern shm_mq_result shm_mq_wait_for_attach(shm_mq_handle *mqh);
+
+/* Smallest possible queue. */
+extern Size shm_mq_minimum_size;
+
+#endif   /* SHM_MQ_H */

commit 084c687250980cb6083fa2dc5367c9429d6a805e
Author: Robert Haas <rhaas@postgresql.org>
Date:   Mon Oct 28 13:06:53 2013 -0400

    Test code for shared memory message queue facility.
    
    This code is intended as a demonstration of how the dynamic shared
    memory and dynamic background worker facilities can be used to establish
    a group of coooperating processes which can coordinate their activities
    using the shared memory message queue facility.  By itself, the code
    does nothing particularly interesting: it simply allows messages to
    be passed through a loop of workers and back to the original process.
    But it's a useful unit test, in addition to its demonstration value.

diff --git a/contrib/Makefile b/contrib/Makefile
index 8a2a937..6e4cfd4 100644
--- a/contrib/Makefile
+++ b/contrib/Makefile
@@ -50,6 +50,7 @@ SUBDIRS = \
 		tablefunc	\
 		tcn		\
 		test_parser	\
+		test_shm_mq	\
 		tsearch2	\
 		unaccent	\
 		vacuumlo	\
diff --git a/contrib/test_shm_mq/.gitignore b/contrib/test_shm_mq/.gitignore
new file mode 100644
index 0000000..5dcb3ff
--- /dev/null
+++ b/contrib/test_shm_mq/.gitignore
@@ -0,0 +1,4 @@
+# Generated subdirectories
+/log/
+/results/
+/tmp_check/
diff --git a/contrib/test_shm_mq/Makefile b/contrib/test_shm_mq/Makefile
new file mode 100644
index 0000000..5e5ac1c
--- /dev/null
+++ b/contrib/test_shm_mq/Makefile
@@ -0,0 +1,20 @@
+# contrib/test_shm_mq/Makefile
+
+MODULE_big = test_shm_mq
+OBJS = test.o setup.o worker.o
+
+EXTENSION = test_shm_mq
+DATA = test_shm_mq--1.0.sql
+
+REGRESS = test_shm_mq
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/test_shm_mq
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/test_shm_mq/expected/test_shm_mq.out b/contrib/test_shm_mq/expected/test_shm_mq.out
new file mode 100644
index 0000000..dd9891c
--- /dev/null
+++ b/contrib/test_shm_mq/expected/test_shm_mq.out
@@ -0,0 +1,18 @@
+CREATE EXTENSION test_shm_mq;
+--
+-- These tests don't produce any interesting output.  We're checking that
+-- the operations complete without crashing or hanging and that none of their
+-- internal sanity tests fail.
+--
+SELECT test_shm_mq(32768, (select string_agg(chr(32+(random()*96)::int), '') from generate_series(1,400)), 10000, 1);
+ test_shm_mq 
+-------------
+ 
+(1 row)
+
+SELECT test_shm_mq_pipelined(16384, (select string_agg(chr(32+(random()*96)::int), '') from generate_series(1,270000)), 200, 3);
+ test_shm_mq_pipelined 
+-----------------------
+ 
+(1 row)
+
diff --git a/contrib/test_shm_mq/setup.c b/contrib/test_shm_mq/setup.c
new file mode 100644
index 0000000..ac1f28c
--- /dev/null
+++ b/contrib/test_shm_mq/setup.c
@@ -0,0 +1,323 @@
+/*--------------------------------------------------------------------------
+ *
+ * setup.c
+ *		Code to set up a dynamic shared memory segments and a specified
+ *		number of background workers for shared memory message queue
+ *		testing.
+ *
+ * Copyright (C) 2013, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/test_shm_mq/setup.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/procsignal.h"
+#include "storage/shm_toc.h"
+#include "utils/memutils.h"
+
+#include "test_shm_mq.h"
+
+typedef struct
+{
+	int		nworkers;
+	BackgroundWorkerHandle *handle[FLEXIBLE_ARRAY_MEMBER];
+} worker_state;
+
+static void setup_dynamic_shared_memory(uint64 queue_size, int nworkers,
+							dsm_segment **segp,
+							test_shm_mq_header **hdrp,
+							shm_mq **outp, shm_mq **inp);
+static worker_state *setup_background_workers(int nworkers,
+											  dsm_segment *seg);
+static void cleanup_background_workers(dsm_segment *seg, Datum arg);
+static void wait_for_workers_to_become_ready(worker_state *wstate,
+								 volatile test_shm_mq_header *hdr);
+static bool check_worker_status(worker_state *wstate);
+
+/*
+ * Set up a dynamic shared memory segment and zero or more background workers
+ * for a test run.
+ */
+void
+test_shm_mq_setup(uint64 queue_size, int32 nworkers, dsm_segment **segp,
+				  shm_mq_handle **output, shm_mq_handle **input)
+{
+	dsm_segment *seg;
+	test_shm_mq_header *hdr;
+	shm_mq	   *outq;
+	shm_mq	   *inq;
+	worker_state	   *wstate;
+
+	/* Set up a dynamic shared memory segment. */
+	setup_dynamic_shared_memory(queue_size, nworkers, &seg, &hdr, &outq, &inq);
+	*segp = seg;
+
+	/* Register background workers. */
+	wstate = setup_background_workers(nworkers, seg);
+
+	/* Attach the queues. */
+	*output = shm_mq_attach(outq, seg, wstate->handle[0]);
+	*input = shm_mq_attach(inq, seg, wstate->handle[nworkers - 1]);
+
+	/* Wait for workers to become ready. */
+	wait_for_workers_to_become_ready(wstate, hdr);
+
+	/*
+	 * Once we reach this point, all workers are ready.  We no longer need
+	 * to kill them if we die; they'll die on their own as the message queues
+	 * shut down.
+	 */
+	cancel_on_dsm_detach(seg, cleanup_background_workers,
+						 PointerGetDatum(wstate));
+	pfree(wstate);
+}
+
+/*
+ * Set up a dynamic shared memory segment.
+ *
+ * We set up a small control region that contains only a test_shm_mq_header,
+ * plus one region per message queue.  There are as many message queues as
+ * the number of workers, plus one.
+ */
+static void
+setup_dynamic_shared_memory(uint64 queue_size, int nworkers,
+							dsm_segment **segp, test_shm_mq_header **hdrp,
+							shm_mq **outp, shm_mq **inp)
+{
+	shm_toc_estimator	e;
+	int					i;
+	uint64			segsize;
+	dsm_segment	   *seg;
+	shm_toc		   *toc;
+	test_shm_mq_header *hdr;
+
+	/* Ensure a valid queue size. */
+	if (queue_size < 0 || ((uint64) queue_size) < shm_mq_minimum_size)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("queue size must be at least " UINT64_FORMAT " bytes",
+					shm_mq_minimum_size)));
+
+	/*
+	 * Estimate how much shared memory we need.
+	 *
+	 * Because the TOC machinery may choose to insert padding of oddly-sized
+	 * requests, we must estimate each chunk separately.
+	 *
+	 * We need one key to register the location of the header, and we need
+	 * nworkers + 1 keys to track the locations of the message queues.
+	 */
+	shm_toc_initialize_estimator(&e);
+	shm_toc_estimate_chunk(&e, sizeof(test_shm_mq_header));
+	for (i = 0; i <= nworkers; ++i)
+		shm_toc_estimate_chunk(&e, queue_size);
+	shm_toc_estimate_keys(&e, 2 + nworkers);
+	segsize = shm_toc_estimate(&e);
+
+	/* Create the shared memory segment and establish a table of contents. */
+	seg = dsm_create(shm_toc_estimate(&e));
+	toc = shm_toc_create(PG_TEST_SHM_MQ_MAGIC, dsm_segment_address(seg),
+						 segsize);
+
+	/* Set up the header region. */
+	hdr = shm_toc_allocate(toc, sizeof(test_shm_mq_header));
+	SpinLockInit(&hdr->mutex);
+	hdr->workers_total = nworkers;
+	hdr->workers_attached = 0;
+	hdr->workers_ready = 0;
+	shm_toc_insert(toc, 0, hdr);
+
+	/* Set up one message queue per worker, plus one. */
+	for (i = 0; i <= nworkers; ++i)
+	{
+		shm_mq		   *mq;
+
+		mq = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
+		shm_toc_insert(toc, i + 1, mq);
+
+		if (i == 0)
+		{
+			/* We send messages to the first queue. */
+			shm_mq_set_sender(mq, MyProc);
+			*outp = mq;
+		}
+		if (i == nworkers)
+		{
+			/* We receive messages from the last queue. */
+			shm_mq_set_receiver(mq, MyProc);
+			*inp = mq;
+		}
+	}
+
+	/* Return results to caller. */
+	*segp = seg;
+	*hdrp = hdr;
+}
+
+/*
+ * Register background workers.
+ */
+static worker_state *
+setup_background_workers(int nworkers, dsm_segment *seg)
+{
+	MemoryContext	oldcontext;
+	BackgroundWorker worker;
+	worker_state	*wstate;
+	int		i;
+
+	/*
+	 * We need the worker_state object and the background worker handles to
+	 * which it points to be allocated in CurTransactionContext rather than
+	 * ExprContext; otherwise, they'll be destroyed before the on_dsm_detach
+	 * hooks run.
+	 */
+	oldcontext = MemoryContextSwitchTo(CurTransactionContext);
+
+	/* Create worker state object. */
+	wstate = MemoryContextAlloc(TopTransactionContext,
+								offsetof(worker_state, handle) +
+								sizeof(BackgroundWorkerHandle *) * nworkers);
+	wstate->nworkers = 0;
+
+	/*
+	 * Arrange to kill all the workers if we abort before all workers are
+	 * finished hooking themselves up to the dynamic shared memory segment.
+	 *
+	 * If we die after all the workers have finished hooking themselves up
+	 * to the dynamic shared memory segment, we'll mark the two queues to
+	 * which we're directly connected as detached, and the worker(s)
+	 * connected to those queues will exit, marking any other queues to
+	 * which they are connected as detached.  This will cause any
+	 * as-yet-unaware workers connected to those queues to exit in their
+	 * turn, and so on, until everybody exits.
+	 *
+	 * But suppose the workers which are supposed to connect to the queues
+	 * to which we're directly attached exit due to some error before they
+	 * actually attach the queues.  The remaining workers will have no way of
+	 * knowing this.  From their perspective, they're still waiting for those
+	 * workers to start, when in fact they've already died.
+	 */
+	on_dsm_detach(seg, cleanup_background_workers,
+				  PointerGetDatum(wstate));
+
+	/* Configure a worker. */
+	worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
+	worker.bgw_start_time = BgWorkerStart_ConsistentState;
+	worker.bgw_restart_time = BGW_NEVER_RESTART;
+	worker.bgw_main = NULL;		/* new worker might not have library loaded */
+	sprintf(worker.bgw_library_name, "test_shm_mq");
+	sprintf(worker.bgw_function_name, "test_shm_mq_main");
+	snprintf(worker.bgw_name, BGW_MAXLEN, "test_shm_mq");
+	worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg));
+	/* set bgw_notify_pid, so we can detect if the worker stops */
+	worker.bgw_notify_pid = MyProcPid;
+
+	/* Register the workers. */
+	for (i = 0; i < nworkers; ++i)
+	{
+		if (!RegisterDynamicBackgroundWorker(&worker, &wstate->handle[i]))
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+					 errmsg("could not register background process"),
+				 errhint("You may need to increase max_worker_processes.")));
+		++wstate->nworkers;
+	}
+
+	/* All done. */
+	MemoryContextSwitchTo(oldcontext);
+	return wstate;
+}
+
+static void
+cleanup_background_workers(dsm_segment *seg, Datum arg)
+{
+	worker_state *wstate = (worker_state *) DatumGetPointer(arg);
+
+	while (wstate->nworkers > 0)
+	{
+		--wstate->nworkers;
+		TerminateBackgroundWorker(wstate->handle[wstate->nworkers]);
+	}
+}
+
+static void
+wait_for_workers_to_become_ready(worker_state *wstate,
+								 volatile test_shm_mq_header *hdr)
+{
+	bool	save_set_latch_on_sigusr1;
+	bool	result = false;
+
+	save_set_latch_on_sigusr1 = set_latch_on_sigusr1;
+	set_latch_on_sigusr1 = true;
+
+	PG_TRY();
+	{
+		for (;;)
+		{
+			int workers_ready;
+
+			/* If all the workers are ready, we have succeeded. */
+			SpinLockAcquire(&hdr->mutex);
+			workers_ready = hdr->workers_ready;
+			SpinLockRelease(&hdr->mutex);
+			if (workers_ready >= wstate->nworkers)
+			{
+				result = true;
+				break;
+			}
+
+			/* If any workers (or the postmaster) have died, we have failed. */
+			if (!check_worker_status(wstate))
+			{
+				result = false;
+				break;
+			}
+
+  			/* Wait to be signalled. */
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+			/* An interrupt may have occurred while we were waiting. */
+			CHECK_FOR_INTERRUPTS();
+
+			/* Reset the latch so we don't spin. */
+			ResetLatch(&MyProc->procLatch);
+		}
+	}
+	PG_CATCH();
+	{
+		set_latch_on_sigusr1 = save_set_latch_on_sigusr1;
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
+
+	if (!result)
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+				 errmsg("one or more background workers failed to start")));
+}
+
+static bool
+check_worker_status(worker_state *wstate)
+{
+	int	n;
+
+	/* If any workers (or the postmaster) have died, we have failed. */
+	for (n = 0; n < wstate->nworkers; ++n)
+	{
+		BgwHandleStatus status;
+		pid_t	pid;
+
+		status = GetBackgroundWorkerPid(wstate->handle[n], &pid);
+		if (status == BGWH_STOPPED || status == BGWH_POSTMASTER_DIED)
+			return false;
+	}
+
+	/* Otherwise, things still look OK. */
+	return true;
+}
diff --git a/contrib/test_shm_mq/sql/test_shm_mq.sql b/contrib/test_shm_mq/sql/test_shm_mq.sql
new file mode 100644
index 0000000..1366df1
--- /dev/null
+++ b/contrib/test_shm_mq/sql/test_shm_mq.sql
@@ -0,0 +1,9 @@
+CREATE EXTENSION test_shm_mq;
+
+--
+-- These tests don't produce any interesting output.  We're checking that
+-- the operations complete without crashing or hanging and that none of their
+-- internal sanity tests fail.
+--
+SELECT test_shm_mq(32768, (select string_agg(chr(32+(random()*96)::int), '') from generate_series(1,400)), 10000, 1);
+SELECT test_shm_mq_pipelined(16384, (select string_agg(chr(32+(random()*96)::int), '') from generate_series(1,270000)), 200, 3);
diff --git a/contrib/test_shm_mq/test.c b/contrib/test_shm_mq/test.c
new file mode 100644
index 0000000..59f18ec
--- /dev/null
+++ b/contrib/test_shm_mq/test.c
@@ -0,0 +1,265 @@
+/*--------------------------------------------------------------------------
+ *
+ * test.c
+ *		Test harness code for shared memory message queues.
+ *
+ * Copyright (C) 2013, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/test_shm_mq/test.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "fmgr.h"
+#include "miscadmin.h"
+
+#include "test_shm_mq.h"
+
+PG_MODULE_MAGIC;
+PG_FUNCTION_INFO_V1(test_shm_mq);
+PG_FUNCTION_INFO_V1(test_shm_mq_pipelined);
+
+void		_PG_init(void);
+Datum		test_shm_mq(PG_FUNCTION_ARGS);
+Datum		test_shm_mq_pipelined(PG_FUNCTION_ARGS);
+
+static void verify_message(uint64 origlen, char *origdata, uint64 newlen,
+			   char *newdata);
+
+/*
+ * Simple test of the shared memory message queue infrastructure.
+ *
+ * We set up a ring of message queues passing through 1 or more background
+ * processes and eventually looping back to ourselves.  We then send a message
+ * through the ring a number of times indicated by the loop count.  At the end,
+ * we check whether the final message matches the one we started with.
+ */
+Datum
+test_shm_mq(PG_FUNCTION_ARGS)
+{
+	int64		queue_size = PG_GETARG_INT64(0);
+	text	   *message = PG_GETARG_TEXT_PP(1);
+	char	   *message_contents = VARDATA_ANY(message);
+	int			message_size = VARSIZE_ANY_EXHDR(message);
+	int32		loop_count = PG_GETARG_INT32(2);
+	int32		nworkers = PG_GETARG_INT32(3);
+	dsm_segment *seg;
+	shm_mq_handle *outqh;
+	shm_mq_handle *inqh;
+	shm_mq_result	res;
+	uint64		len;
+	void	   *data;
+
+	/* A negative loopcount is nonsensical. */
+	if (loop_count < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("repeat count size must be a non-negative integer")));
+
+	/*
+	 * Since this test sends data using the blocking interfaces, it cannot
+	 * send data to itself.  Therefore, a minimum of 1 worker is required.
+	 * Of course, a negative worker count is nonsensical.
+	 */
+	if (nworkers < 1)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("number of workers must be a positive integer")));
+
+	/* Set up dynamic shared memory segment and background workers. */
+	test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
+
+	/* Send the initial message. */
+	res = shm_mq_send(outqh, message_size, message_contents, false);
+	if (res != SHM_MQ_SUCCESS)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("could not send message")));
+
+	/*
+	 * Receive a message and send it back out again.  Do this a number of
+	 * times equal to the loop count.
+	 */
+	for (;;)
+	{
+		/* Receive a message. */
+		res = shm_mq_receive(inqh, &len, &data, false);
+		if (res != SHM_MQ_SUCCESS)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("could not receive message")));
+
+		/* If this is supposed to be the last iteration, stop here. */
+		if (--loop_count <= 0)
+			break;
+
+		/* Send it back out. */
+		res = shm_mq_send(outqh, len, data, false);
+		if (res != SHM_MQ_SUCCESS)
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("could not send message")));
+	}
+
+	/*
+	 * Finally, check that we got back the same message from the last
+	 * iteration that we originally sent.
+	 */
+	verify_message(message_size, message_contents, len, data);
+
+	/* Clean up. */
+	dsm_detach(seg);
+
+	PG_RETURN_VOID();
+}
+
+/*
+ * Pipelined test of the shared memory message queue infrastructure.
+ *
+ * As in the basic test, we set up a ring of message queues passing through
+ * 1 or more background processes and eventually looping back to ourselves.
+ * Then, we send N copies of the user-specified message through the ring and
+ * receive them all back.  Since this might fill up all message queues in the
+ * ring and then stall, we must be prepared to begin receiving the messages
+ * back before we've finished sending them.
+ */
+Datum
+test_shm_mq_pipelined(PG_FUNCTION_ARGS)
+{
+	int64		queue_size = PG_GETARG_INT64(0);
+	text	   *message = PG_GETARG_TEXT_PP(1);
+	char	   *message_contents = VARDATA_ANY(message);
+	int			message_size = VARSIZE_ANY_EXHDR(message);
+	int32		loop_count = PG_GETARG_INT32(2);
+	int32		nworkers = PG_GETARG_INT32(3);
+	bool		verify = PG_GETARG_BOOL(4);
+	int32		send_count = 0;
+	int32		receive_count = 0;
+	dsm_segment *seg;
+	shm_mq_handle *outqh;
+	shm_mq_handle *inqh;
+	shm_mq_result	res;
+	uint64		len;
+	void	   *data;
+
+	/* A negative loopcount is nonsensical. */
+	if (loop_count < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("repeat count size must be a non-negative integer")));
+
+	/*
+	 * Using the nonblocking interfaces, we can even send data to ourselves,
+	 * so the minimum number of workers for this test is zero.
+	 */
+	if (nworkers < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("number of workers must be a non-negative integer")));
+
+	/* Set up dynamic shared memory segment and background workers. */
+	test_shm_mq_setup(queue_size, nworkers, &seg, &outqh, &inqh);
+
+	/* Main loop. */
+	for (;;)
+	{
+		bool		wait = true;
+
+		/*
+		 * If we haven't yet sent the message the requisite number of times,
+		 * try again to send it now.  Note that when shm_mq_send() returns
+		 * SHM_MQ_WOULD_BLOCK, the next call to that function must pass the
+		 * same message size and contents; that's not an issue here because
+		 * we're sending the same message every time.
+		 */
+		if (send_count < loop_count)
+		{
+			res = shm_mq_send(outqh, message_size, message_contents, true);
+			if (res == SHM_MQ_SUCCESS)
+			{
+				++send_count;
+				wait = false;
+			}
+			else if (res == SHM_MQ_DETACHED)
+				ereport(ERROR,
+						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+						 errmsg("could not send message")));
+		}
+
+		/*
+		 * If we haven't yet received the message the requisite number of
+		 * times, try to receive it again now.
+		 */
+		if (receive_count < loop_count)
+		{
+			res = shm_mq_receive(inqh, &len, &data, true);
+			if (res == SHM_MQ_SUCCESS)
+			{
+				++receive_count;
+				/* Verifying every time is slow, so it's optional. */
+				if (verify)
+					verify_message(message_size, message_contents, len, data);
+				wait = false;
+			}
+			else if (res == SHM_MQ_DETACHED)
+				ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("could not receive message")));
+		}
+		else
+		{
+			/*
+			 * Otherwise, we've received the message enough times.  This
+			 * shouldn't happen unless we've also sent it enough times.
+			 */
+			if (send_count != receive_count)
+				ereport(ERROR,
+						(errcode(ERRCODE_INTERNAL_ERROR),
+						 errmsg("message sent %d times, but received %d times",
+							send_count, receive_count)));
+			break;
+		}
+
+		if (wait)
+		{
+			/*
+			 * If we made no progress, wait for one of the other processes
+			 * to which we are connected to set our latch, indicating that
+			 * they have read or written data and therefore there may now be
+			 * work for us to do.
+			 */
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+			CHECK_FOR_INTERRUPTS();
+			ResetLatch(&MyProc->procLatch);
+		}
+	}
+
+	/* Clean up. */
+	dsm_detach(seg);
+
+	PG_RETURN_VOID();
+}
+
+/*
+ * Verify that two messages are the same.
+ */
+static void
+verify_message(uint64 origlen, char *origdata, uint64 newlen, char *newdata)
+{
+	uint64	i;
+
+	if (origlen != newlen)
+		ereport(ERROR,
+				(errmsg("message corrupted"),
+				 errdetail("The original message was " UINT64_FORMAT " bytes but the final message is " UINT64_FORMAT " bytes.",
+					 origlen, newlen)));
+
+	for (i = 0; i < origlen; ++i)
+		if (origdata[i] != newdata[i])
+			ereport(ERROR,
+					(errmsg("message corrupted"),
+					 errdetail("The new and original messages differ at byte " UINT64_FORMAT " of " UINT64_FORMAT ".", i, origlen)));
+}
diff --git a/contrib/test_shm_mq/test_shm_mq--1.0.sql b/contrib/test_shm_mq/test_shm_mq--1.0.sql
new file mode 100644
index 0000000..54b225e
--- /dev/null
+++ b/contrib/test_shm_mq/test_shm_mq--1.0.sql
@@ -0,0 +1,19 @@
+/* contrib/test_shm_mq/test_shm_mq--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION test_shm_mq" to load this file. \quit
+
+CREATE FUNCTION test_shm_mq(queue_size pg_catalog.int8,
+					   message pg_catalog.text,
+					   repeat_count pg_catalog.int4 default 1,
+					   num_workers pg_catalog.int4 default 1)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
+
+CREATE FUNCTION test_shm_mq_pipelined(queue_size pg_catalog.int8,
+					   message pg_catalog.text,
+					   repeat_count pg_catalog.int4 default 1,
+					   num_workers pg_catalog.int4 default 1,
+					   verify pg_catalog.bool default true)
+    RETURNS pg_catalog.void STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
diff --git a/contrib/test_shm_mq/test_shm_mq.control b/contrib/test_shm_mq/test_shm_mq.control
new file mode 100644
index 0000000..d9a74c7
--- /dev/null
+++ b/contrib/test_shm_mq/test_shm_mq.control
@@ -0,0 +1,4 @@
+comment = 'Test code for shared memory message queues'
+default_version = '1.0'
+module_pathname = '$libdir/test_shm_mq'
+relocatable = true
diff --git a/contrib/test_shm_mq/test_shm_mq.h b/contrib/test_shm_mq/test_shm_mq.h
new file mode 100644
index 0000000..5f87da7
--- /dev/null
+++ b/contrib/test_shm_mq/test_shm_mq.h
@@ -0,0 +1,45 @@
+/*--------------------------------------------------------------------------
+ *
+ * test_shm_mq.h
+ *		Definitions for shared memory message queues
+ *
+ * Copyright (C) 2013, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/test_shm_mq/test_shm_mq.h
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#ifndef TEST_SHM_MQ_H
+#define TEST_SHM_MQ_H
+
+#include "storage/dsm.h"
+#include "storage/shm_mq.h"
+#include "storage/spin.h"
+
+/* Identifier for shared memory segments used by this extension. */
+#define		PG_TEST_SHM_MQ_MAGIC		0x79fb2447
+
+/*
+ * This structure is stored in the dynamic shared memory segment.  We use
+ * it to determine whether all workers started up OK and successfully
+ * attached to their respective shared message queues.
+ */
+typedef struct
+{
+	slock_t			mutex;
+	int				workers_total;
+	int				workers_attached;
+	int				workers_ready;
+} test_shm_mq_header;
+
+/* Set up dynamic shared memory and background workers for test run. */
+extern void test_shm_mq_setup(uint64 queue_size, int32 nworkers,
+							  dsm_segment **seg, shm_mq_handle **output,
+							  shm_mq_handle **input);
+
+/* Main entrypoint for a worker. */
+extern void	test_shm_mq_main(Datum);
+
+#endif
diff --git a/contrib/test_shm_mq/worker.c b/contrib/test_shm_mq/worker.c
new file mode 100644
index 0000000..95b23c9
--- /dev/null
+++ b/contrib/test_shm_mq/worker.c
@@ -0,0 +1,224 @@
+/*--------------------------------------------------------------------------
+ *
+ * worker.c
+ *		Code for sample worker making use of shared memory message queues.
+ *		Our test worker simply reads messages from one message queue and
+ *		writes them back out to another message queue.  In a real
+ *		application, you'd presumably want the worker to do some more
+ *		complex calculation rather than simply returning the input,
+ *		but it should be possible to use much of the control logic just
+ *		as presented here.
+ *
+ * Copyright (C) 2013, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/test_shm_mq/worker.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "storage/ipc.h"
+#include "storage/procarray.h"
+#include "storage/shm_mq.h"
+#include "storage/shm_toc.h"
+#include "utils/resowner.h"
+
+#include "test_shm_mq.h"
+
+static void handle_sigterm(SIGNAL_ARGS);
+static void attach_to_queues(dsm_segment *seg, shm_toc *toc,
+							 int myworkernumber, shm_mq_handle **inqhp,
+							 shm_mq_handle **outqhp);
+static void copy_messages(shm_mq_handle *inqh, shm_mq_handle *outqh);
+
+/*
+ * Background worker entrypoint.
+ *
+ * This is intended to demonstrate how a background worker can be used to
+ * facilitate a parallel computation.  Most of the logic here is fairly
+ * boilerplate stuff, designed to attach to the shared memory segment,
+ * notify the user backend that we're alive, and so on.  The
+ * application-specific bits of logic that you'd replace for your own worker
+ * are attach_to_queues() and copy_messages().
+ */
+void
+test_shm_mq_main(Datum main_arg)
+{
+	dsm_segment *seg;
+	shm_toc	   *toc;
+	shm_mq_handle *inqh;
+	shm_mq_handle *outqh;
+	volatile test_shm_mq_header *hdr;
+	int			myworkernumber;
+	PGPROC	   *registrant;
+
+	/*
+	 * Establish signal handlers.
+	 *
+	 * We want CHECK_FOR_INTERRUPTS() to kill off this worker process just
+	 * as it would a normal user backend.  To make that happen, we establish
+	 * a signal handler that is a stripped-down version of die().  We don't
+	 * have any equivalent of the backend's command-read loop, where interrupts
+	 * can be processed immediately, so make sure ImmediateInterruptOK is
+	 * turned off.
+	 */
+	pqsignal(SIGTERM, handle_sigterm);
+	ImmediateInterruptOK = false;
+	BackgroundWorkerUnblockSignals();
+
+	/*
+	 * Connect to the dynamic shared memory segment.
+	 *
+	 * The backend that registered this worker passed us the ID of a shared
+	 * memory segment to which we must attach for further instructions.  In
+	 * order to attach to dynamic shared memory, we need a resource owner.
+	 * Once we've mapped the segment in our address space, attach to the table
+	 * of contents so we can locate the various data structures we'll need
+	 * to find within the segment.
+	 */
+	CurrentResourceOwner = ResourceOwnerCreate(NULL, "test_shm_mq worker");
+	seg = dsm_attach(DatumGetInt32(main_arg));
+	if (seg == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("unable to map dynamic shared memory segment")));
+	toc = shm_toc_attach(PG_TEST_SHM_MQ_MAGIC, dsm_segment_address(seg));
+	if (toc == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("bad magic number in dynamic shared memory segment")));
+
+	/*
+	 * Acquire a worker number.
+	 *
+	 * By convention, the process registering this background worker should
+	 * have stored the control structure at key 0.  We look up that key to
+	 * find it.  Our worker number gives our identity: there may be just one
+	 * worker involved in this parallel operation, or there may be many.
+	 */
+	hdr = shm_toc_lookup(toc, 0);
+	SpinLockAcquire(&hdr->mutex);
+	myworkernumber = ++hdr->workers_attached;
+	SpinLockRelease(&hdr->mutex);
+	if (myworkernumber > hdr->workers_total)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("too many message queue testing workers already")));
+
+	/*
+	 * Attach to the appropriate message queues.
+	 */
+	attach_to_queues(seg, toc, myworkernumber, &inqh, &outqh);
+
+	/*
+	 * Indicate that we're fully initialized and ready to begin the main
+	 * part of the parallel operation.
+	 *
+	 * Once we signal that we're ready, the user backend is entitled to assume
+	 * that our on_dsm_detach callbacks will fire before we disconnect from
+	 * the shared memory segment and exit.  Generally, that means we must have
+	 * attached to all relevant dynamic shared memory data structures by now.
+	 */
+	SpinLockAcquire(&hdr->mutex);
+	++hdr->workers_ready;
+	SpinLockRelease(&hdr->mutex);
+	registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
+	if (registrant == NULL)		
+	{
+		elog(DEBUG1, "registrant backend has exited prematurely");
+		proc_exit(1);
+	}
+	SetLatch(&registrant->procLatch);
+
+	/* Do the work. */
+	copy_messages(inqh, outqh);
+
+	/*
+	 * We're done.  Explicitly detach the shared memory segment so that we
+	 * don't get a resource leak warning at commit time.  This will fire any
+	 * on_dsm_detach callbacks we've registered, as well.  Once that's done,
+	 * we can go ahead and exit.
+	 */
+	dsm_detach(seg);
+	proc_exit(1);
+}
+
+/*
+ * Attach to shared memory message queues.
+ *
+ * We use our worker number to determine to which queue we should attach.
+ * The queues are registered at keys 1..<number-of-workers>.  The user backend
+ * writes to queue #1 and reads from queue #<number-of-workers>; each worker
+ * reads from the queue whose number is equal to its worker number and writes
+ * to the next higher-numbered queue.
+ */
+static void
+attach_to_queues(dsm_segment *seg, shm_toc *toc, int myworkernumber,
+				 shm_mq_handle **inqhp, shm_mq_handle **outqhp)
+{
+	shm_mq	   *inq;
+	shm_mq	   *outq;
+
+	inq = shm_toc_lookup(toc, myworkernumber);
+	shm_mq_set_receiver(inq, MyProc);
+	*inqhp = shm_mq_attach(inq, seg, NULL);
+	outq = shm_toc_lookup(toc, myworkernumber + 1);
+	shm_mq_set_sender(outq, MyProc);
+	*outqhp = shm_mq_attach(outq, seg, NULL);
+}
+
+/*
+ * Loop, receiving and sending messages, until the connection is broken.
+ *
+ * This is the "real work" performed by this worker process.  Everything that
+ * happens before this is initialization of one form or another, and everything
+ * after this point is cleanup.
+ */
+static void
+copy_messages(shm_mq_handle *inqh, shm_mq_handle *outqh)
+{
+	uint64		len;
+	void	   *data;
+	shm_mq_result res;
+
+	for (;;)
+	{
+		/* Notice any interrupts that have occurred. */
+		CHECK_FOR_INTERRUPTS();
+
+		/* Receive a message. */
+		res = shm_mq_receive(inqh, &len, &data, false);
+		if (res != SHM_MQ_SUCCESS)
+			break;
+
+		/* Send it back out. */
+		res = shm_mq_send(outqh, len, data, false);
+		if (res != SHM_MQ_SUCCESS)
+			break;
+	}
+}
+
+/*
+ * When we receive a SIGTERM, we set InterruptPending and ProcDiePending just
+ * like a normal backend.  The next CHECK_FOR_INTERRUPTS() will do the right
+ * thing.
+ */
+static void
+handle_sigterm(SIGNAL_ARGS)
+{
+	int			save_errno = errno;
+
+	if (MyProc)
+		SetLatch(&MyProc->procLatch);
+
+	if (!proc_exit_inprogress)
+	{
+		InterruptPending = true;
+		ProcDiePending = true;
+	}
+
+	errno = save_errno;
+}

diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
index 26d7b7c..2d298a3 100644
--- a/src/backend/storage/ipc/shm_mq.c
+++ b/src/backend/storage/ipc/shm_mq.c
@@ -25,6 +25,8 @@
 #include "storage/spin.h"
 
 /*
+ * This structure represents the actual queue, stored in shared memory.
+ *
  * Some notes on synchronization:
  *
  * mq_receiver and mq_bytes_read can only be changed by the receiver; and
@@ -77,6 +79,8 @@ struct shm_mq
 };
 
 /*
+ * This structure is a backend-private handle for access to a queue.
+ *
  * mqh_queue is a pointer to the queue we've attached, and mqh_segment is
  * a pointer to the dynamic shared memory segment that contains it.
  *
@@ -147,10 +151,10 @@ static shm_mq_result shm_mq_notify_receiver(volatile shm_mq *mq);
 static void shm_mq_detach_callback(dsm_segment *seg, Datum arg);
 
 /* Minimum queue size is enough for header and at least one chunk of data. */
-Size shm_mq_minimum_size =
+const Size shm_mq_minimum_size =
 	MAXALIGN(offsetof(shm_mq, mq_ring)) + MAXIMUM_ALIGNOF;
 
-#define MQH_BUFSIZE				8192
+#define MQH_INITIAL_BUFSIZE				8192
 
 /*
  * Initialize a new shared message queue.
@@ -168,6 +172,7 @@ shm_mq_create(void *address, Size size)
 	Assert(size > data_offset);
 
 	/* Initialize queue header. */
+	SpinLockInit(&mq->mq_mutex);
 	mq->mq_receiver = NULL;
 	mq->mq_sender = NULL;
 	mq->mq_bytes_read = 0;
@@ -255,6 +260,8 @@ shm_mq_get_sender(shm_mq *mq)
  *
  * The memory context in effect at the time this function is called should
  * be one which will last for at least as long as the message queue itself.
+ * We'll allocate the handle in that context, and future allocations that
+ * are needed to buffer incoming data will happen in that context as well.
  *
  * If seg != NULL, the queue will be automatically detached when that dynamic
  * shared memory segment is detached.
@@ -302,7 +309,9 @@ shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
  * When nowait = true, we do not manipulate the state of the process latch;
  * instead, if the buffer becomes full, we return SHM_MQ_WOULD_BLOCK.  In
  * this case, the caller should call this function again, with the same
- * arguments, each time the process latch is set.
+ * arguments, each time the process latch is set.  (Once begun, the sending
+ * of a message cannot be aborted except by detaching from the queue; changing
+ * the length or payload will corrupt the queue.)
  */
 shm_mq_result
 shm_mq_send(shm_mq_handle *mqh, uint64 nbytes, void *data, bool nowait)
@@ -468,7 +477,7 @@ shm_mq_receive(shm_mq_handle *mqh, uint64 *nbytesp, void **datap, bool nowait)
 		 */
 		if (mqh->mqh_buflen < nbytes)
 		{
-			uint64		newbuflen = Max(mqh->mqh_buflen, MQH_BUFSIZE);
+			uint64		newbuflen = Max(mqh->mqh_buflen, MQH_INITIAL_BUFSIZE);
 
 			while (newbuflen < nbytes)
 				newbuflen *= 2;
@@ -856,7 +865,7 @@ shm_mq_get_bytes_read(volatile shm_mq *mq, bool *detached)
 	*detached = mq->mq_detached;
 	SpinLockRelease(&mq->mq_mutex);
 
-	return mq->mq_bytes_read;
+	return v;
 }
 
 /*
@@ -891,7 +900,7 @@ shm_mq_get_bytes_written(volatile shm_mq *mq, bool *detached)
 	*detached = mq->mq_detached;
 	SpinLockRelease(&mq->mq_mutex);
 
-	return mq->mq_bytes_written;
+	return v;
 }
 
 /*
diff --git a/src/include/storage/shm_mq.h b/src/include/storage/shm_mq.h
index 749ab3b..1ce88a1 100644
--- a/src/include/storage/shm_mq.h
+++ b/src/include/storage/shm_mq.h
@@ -65,6 +65,6 @@ extern shm_mq_result shm_mq_receive(shm_mq_handle *mqh,
 extern shm_mq_result shm_mq_wait_for_attach(shm_mq_handle *mqh);
 
 /* Smallest possible queue. */
-extern Size shm_mq_minimum_size;
+extern const Size shm_mq_minimum_size;
 
 #endif   /* SHM_MQ_H */

diff --git a/src/backend/storage/ipc/Makefile b/src/backend/storage/ipc/Makefile
index df0a49e..850347c 100644
--- a/src/backend/storage/ipc/Makefile
+++ b/src/backend/storage/ipc/Makefile
@@ -16,6 +16,6 @@ endif
 endif
 
 OBJS = dsm_impl.o dsm.o ipc.o ipci.o pmsignal.o procarray.o procsignal.o \
-	shmem.o shmqueue.o shm_toc.o sinval.o sinvaladt.o standby.o
+	shmem.o shmqueue.o shm_mq.o shm_toc.o sinval.o sinvaladt.o standby.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/ipc/shm_mq.c b/src/backend/storage/ipc/shm_mq.c
new file mode 100644
index 0000000..2d298a3
--- /dev/null
+++ b/src/backend/storage/ipc/shm_mq.c
@@ -0,0 +1,945 @@
+/*-------------------------------------------------------------------------
+ *
+ * shm_mq.c
+ *	  single-reader, single-writer shared memory message queue
+ *
+ * Both the sender and the receiver must have a PGPROC; their respective
+ * process latches are used for synchronization.  Only the sender may send,
+ * and only the receiver may receive.  This is intended to allow a user
+ * backend to communicate with worker backends that it has registered.
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/storage/shm_mq.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/procsignal.h"
+#include "storage/shm_mq.h"
+#include "storage/spin.h"
+
+/*
+ * This structure represents the actual queue, stored in shared memory.
+ *
+ * Some notes on synchronization:
+ *
+ * mq_receiver and mq_bytes_read can only be changed by the receiver; and
+ * mq_sender and mq_bytes_written can only be changed by the sender.  However,
+ * because most of these fields are 8 bytes and we don't assume that 8 byte
+ * reads and writes are atomic, the spinlock must be taken whenever the field
+ * is updated, and whenever it is read by a process other than the one allowed
+ * to modify it. But the process that is allowed to modify it is also allowed
+ * to read it without the lock.  On architectures where 8-byte writes are
+ * atomic, we could replace these spinlocks with memory barriers, but
+ * testing found no performance benefit, so it seems best to keep things
+ * simple for now.
+ *
+ * mq_detached can be set by either the sender or the receiver, so the mutex
+ * must be held to read or write it.  Memory barriers could be used here as
+ * well, if needed.
+ *
+ * mq_ring_size and mq_ring_offset never change after initialization, and
+ * can therefore be read without the lock.
+ *
+ * Importantly, mq_ring can be safely read and written without a lock.  Were
+ * this not the case, we'd have to hold the spinlock for much longer
+ * intervals, and performance might suffer.  Fortunately, that's not
+ * necessary.  At any given time, the difference between mq_bytes_read and
+ * mq_bytes_written defines the number of bytes within mq_ring that contain
+ * unread data, and mq_bytes_read defines the position where those bytes
+ * begin.  The sender can increase the number of unread bytes at any time,
+ * but only the receiver can give license to overwrite those bytes, by
+ * incrementing mq_bytes_read.  Therefore, it's safe for the receiver to read
+ * the unread bytes it knows to be present without the lock.  Conversely,
+ * the sender can write to the unused portion of the ring buffer without
+ * the lock, because nobody else can be reading or writing those bytes.  The
+ * receiver could be making more bytes unused by incrementing mq_bytes_read,
+ * but that's OK.  Note that it would be unsafe for the receiver to read any
+ * data it's already marked as read, or to write any data; and it would be
+ * unsafe for the sender to reread any data after incrementing
+ * mq_bytes_written, but fortunately there's no need for any of that.
+ */
+struct shm_mq
+{
+	slock_t		mq_mutex;
+	PGPROC	   *mq_receiver;
+	PGPROC	   *mq_sender;
+	uint64		mq_bytes_read;
+	uint64		mq_bytes_written;
+	uint64		mq_ring_size;
+	bool		mq_detached;
+	uint8		mq_ring_offset;
+	char		mq_ring[FLEXIBLE_ARRAY_MEMBER];
+};
+
+/*
+ * This structure is a backend-private handle for access to a queue.
+ *
+ * mqh_queue is a pointer to the queue we've attached, and mqh_segment is
+ * a pointer to the dynamic shared memory segment that contains it.
+ *
+ * If this queue is intended to connect the current process with a background
+ * worker that started it, the user can pass a pointer to the worker handle
+ * to shm_mq_attach(), and we'll store it in mqh_handle.  The point of this
+ * is to allow us to begin sending to or receiving from that queue before the
+ * process we'll be communicating with has even been started.  If it fails
+ * to start, the handle will allow us to notice that and fail cleanly, rather
+ * than waiting forever; see shm_mq_wait_internal.  This is mostly useful in
+ * simple cases - e.g. where there are just 2 processes communicating; in
+ * more complex scenarios, every process may not have a BackgroundWorkerHandle
+ * available, or may need to watch for the failure of more than one other
+ * process at a time.
+ *
+ * When a message exists as a contiguous chunk of bytes in the queue - that is,
+ * it is smaller than the size of the ring buffer and does not wrap around
+ * the end - we return the message to the caller as a pointer into the buffer.
+ * For messages that are larger or happen to wrap, we reassemble the message
+ * locally by copying the chunks into a backend-local buffer.  mqh_buffer is
+ * the buffer, and mqh_buflen is the number of bytes allocated for it.
+ *
+ * mqh_partial_message_bytes, mqh_expected_bytes, and mqh_did_length_word
+ * are used to track the state of non-blocking operations.  When the caller
+ * attempts a non-blocking operation that returns SHM_MQ_WOULD_BLOCK, they
+ * are expected to retry the call at a later time with the same argument;
+ * we need to retain enough state to pick up where we left off.
+ * mqh_did_length_word tracks whether we read or wrote the length word,
+ * mqh_partial_message_bytes tracks the number of payload bytes read or
+ * written, and mqh_expected_bytes - which is used only for reads - tracks
+ * the expected total size of the payload.
+ *
+ * mqh_counterparty_attached tracks whether we know the counterparty to have
+ * attached to the queue at some previous point.  This lets us avoid some
+ * mutex acquisitions.
+ *
+ * mqh_context is the memory context in effect at the time we attached to
+ * the shm_mq.  The shm_mq_handle itself is allocated in this context, and
+ * we make sure any other allocations we do happen in this context as well,
+ * to avoid nasty surprises.
+ */
+struct shm_mq_handle
+{
+	shm_mq	   *mqh_queue;
+	dsm_segment *mqh_segment;
+	BackgroundWorkerHandle *mqh_handle;
+	char	   *mqh_buffer;
+	uint64		mqh_buflen;
+	uint64		mqh_consume_pending;
+	uint64		mqh_partial_message_bytes;
+	uint64		mqh_expected_bytes;
+	bool		mqh_did_length_word;
+	bool		mqh_counterparty_attached;
+	MemoryContext mqh_context;
+};
+
+static shm_mq_result shm_mq_send_bytes(shm_mq_handle *mq, uint64 nbytes,
+				  void *data, bool nowait, uint64 *bytes_written);
+static shm_mq_result shm_mq_receive_bytes(shm_mq *mq, uint64 bytes_needed,
+					 bool nowait, uint64 *nbytesp, void **datap);
+static bool shm_mq_wait_internal(volatile shm_mq *mq, PGPROC * volatile *ptr,
+					 BackgroundWorkerHandle *handle);
+static uint64 shm_mq_get_bytes_read(volatile shm_mq *mq, bool *detached);
+static void shm_mq_inc_bytes_read(volatile shm_mq *mq, uint64 n);
+static uint64 shm_mq_get_bytes_written(volatile shm_mq *mq, bool *detached);
+static void shm_mq_inc_bytes_written(volatile shm_mq *mq, uint64 n);
+static shm_mq_result shm_mq_notify_receiver(volatile shm_mq *mq);
+static void shm_mq_detach_callback(dsm_segment *seg, Datum arg);
+
+/* Minimum queue size is enough for header and at least one chunk of data. */
+const Size shm_mq_minimum_size =
+	MAXALIGN(offsetof(shm_mq, mq_ring)) + MAXIMUM_ALIGNOF;
+
+#define MQH_INITIAL_BUFSIZE				8192
+
+/*
+ * Initialize a new shared message queue.
+ */
+shm_mq *
+shm_mq_create(void *address, Size size)
+{
+	shm_mq	   *mq = address;
+	uint64		data_offset = MAXALIGN(offsetof(shm_mq, mq_ring));
+
+	/* If the size isn't MAXALIGN'd, just discard the odd bytes. */
+	size = MAXALIGN_DOWN(size);
+
+	/* Queue size must be large enough to hold some data. */
+	Assert(size > data_offset);
+
+	/* Initialize queue header. */
+	SpinLockInit(&mq->mq_mutex);
+	mq->mq_receiver = NULL;
+	mq->mq_sender = NULL;
+	mq->mq_bytes_read = 0;
+	mq->mq_bytes_written = 0;
+	mq->mq_ring_size = size - data_offset;
+	mq->mq_detached = false;
+	mq->mq_ring_offset = data_offset - offsetof(shm_mq, mq_ring);
+
+	return mq;
+}
+
+/*
+ * Set the identity of the process that will receive from a shared message
+ * queue.
+ */
+void
+shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *sender;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	Assert(vmq->mq_receiver == NULL);
+	vmq->mq_receiver = proc;
+	sender = vmq->mq_sender;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (sender != NULL)
+		SetLatch(&sender->procLatch);
+}
+
+/*
+ * Set the identity of the process that will send to a shared message queue.
+ */
+void
+shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *receiver;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	Assert(vmq->mq_sender == NULL);
+	vmq->mq_sender = proc;
+	receiver = vmq->mq_receiver;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (receiver != NULL)
+		SetLatch(&receiver->procLatch);
+}
+
+/*
+ * Get the configured receiver.
+ */
+PGPROC *
+shm_mq_get_receiver(shm_mq *mq)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *receiver;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	receiver = vmq->mq_receiver;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return receiver;
+}
+
+/*
+ * Get the configured sender.
+ */
+PGPROC *
+shm_mq_get_sender(shm_mq *mq)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC   *sender;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	sender = vmq->mq_sender;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return sender;
+}
+
+/*
+ * Attach to a shared message queue so we can send or receive messages.
+ *
+ * The memory context in effect at the time this function is called should
+ * be one which will last for at least as long as the message queue itself.
+ * We'll allocate the handle in that context, and future allocations that
+ * are needed to buffer incoming data will happen in that context as well.
+ *
+ * If seg != NULL, the queue will be automatically detached when that dynamic
+ * shared memory segment is detached.
+ *
+ * If handle != NULL, the queue can be read or written even before the
+ * other process has attached.  We'll wait for it to do so if needed.  The
+ * handle must be for a background worker initialized with bgw_notify_pid
+ * equal to our PID.
+ *
+ * shm_mq_detach() should be called when done.  This will free the
+ * shm_mq_handle and mark the queue itself as detached, so that our
+ * counterpart won't get stuck waiting for us to fill or drain the queue
+ * after we've already lost interest.
+ */
+shm_mq_handle *
+shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
+{
+	shm_mq_handle	   *mqh = palloc(sizeof(shm_mq_handle));
+
+	Assert(mq->mq_receiver == MyProc || mq->mq_sender == MyProc);
+	mqh->mqh_queue = mq;
+	mqh->mqh_segment = seg;
+	mqh->mqh_buffer = NULL;
+	mqh->mqh_handle = handle;
+	mqh->mqh_buflen = 0;
+	mqh->mqh_consume_pending = 0;
+	mqh->mqh_context = CurrentMemoryContext;
+	mqh->mqh_partial_message_bytes = 0;
+	mqh->mqh_did_length_word = false;
+	mqh->mqh_counterparty_attached = false;
+
+	if (seg != NULL)
+		on_dsm_detach(seg, shm_mq_detach_callback, PointerGetDatum(mq));
+
+	return mqh;
+}
+
+/*
+ * Write a message into a shared message queue.
+ *
+ * When nowait = false, we'll wait on our process latch when the ring buffer
+ * fills up, and then continue writing once the receiver has drained some data.
+ * The process latch is reset after each wait.
+ *
+ * When nowait = true, we do not manipulate the state of the process latch;
+ * instead, if the buffer becomes full, we return SHM_MQ_WOULD_BLOCK.  In
+ * this case, the caller should call this function again, with the same
+ * arguments, each time the process latch is set.  (Once begun, the sending
+ * of a message cannot be aborted except by detaching from the queue; changing
+ * the length or payload will corrupt the queue.)
+ */
+shm_mq_result
+shm_mq_send(shm_mq_handle *mqh, uint64 nbytes, void *data, bool nowait)
+{
+	shm_mq_result	res;
+	shm_mq		   *mq = mqh->mqh_queue;
+	uint64			bytes_written;
+
+	Assert(mq->mq_sender == MyProc);
+
+	/* Write the message length into the buffer. */
+	if (!mqh->mqh_did_length_word)
+	{
+		res = shm_mq_send_bytes(mqh, sizeof(uint64), &nbytes, nowait,
+								&bytes_written);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+
+		/*
+		 * We're sure to have sent the length in full, since we always
+		 * write a MAXALIGN'd chunk.
+		 */
+		Assert(bytes_written == MAXALIGN64(sizeof(uint64)));
+		mqh->mqh_did_length_word = true;
+	}
+
+	/* Write the actual data bytes into the buffer. */
+	Assert(mqh->mqh_partial_message_bytes <= nbytes);
+	res = shm_mq_send_bytes(mqh, nbytes - mqh->mqh_partial_message_bytes,
+							((char *) data) + mqh->mqh_partial_message_bytes,
+							nowait, &bytes_written);
+	if (res == SHM_MQ_WOULD_BLOCK)
+		mqh->mqh_partial_message_bytes += bytes_written;
+	else
+	{
+		mqh->mqh_partial_message_bytes = 0;
+		mqh->mqh_did_length_word = false;
+	}
+	if (res != SHM_MQ_SUCCESS)
+		return res;
+
+	/* Notify receiver of the newly-written data, and return. */
+	return shm_mq_notify_receiver(mq);
+}
+
+/*
+ * Receive a message from a shared message queue.
+ *
+ * We set *nbytes to the message length and *data to point to the message
+ * payload.  If the entire message exists in the queue as a single,
+ * contiguous chunk, *data will point directly into shared memory; otherwise,
+ * it will point to a temporary buffer.  This mostly avoids data copying in
+ * the hoped-for case where messages are short compared to the buffer size,
+ * while still allowing longer messages.  In either case, the return value
+ * remains valid until the next receive operation is perfomed on the queue.
+ *
+ * When nowait = false, we'll wait on our process latch when the ring buffer
+ * is empty and we have not yet received a full message.  The sender will
+ * set our process latch after more data has been written, and we'll resume
+ * processing.  Each call will therefore return a complete message
+ * (unless the sender detaches the queue).
+ *
+ * When nowait = true, we do not manipulate the state of the process latch;
+ * instead, whenever the buffer is empty and we need to read from it, we
+ * return SHM_MQ_WOULD_BLOCK.  In this case, the caller should call this
+ * function again after the process latch has been set.
+ */
+shm_mq_result
+shm_mq_receive(shm_mq_handle *mqh, uint64 *nbytesp, void **datap, bool nowait)
+{
+	shm_mq		   *mq = mqh->mqh_queue;
+	shm_mq_result	res;
+	uint64			rb = 0;
+	uint64			nbytes;
+	uint64			needed;
+	void		   *rawdata;
+
+	Assert(mq->mq_receiver == MyProc);
+
+	/* We can't receive data until the sender has attached. */
+	if (!mqh->mqh_counterparty_attached)
+	{
+		if (nowait)
+		{
+			if (shm_mq_get_sender(mq) == NULL)
+				return SHM_MQ_WOULD_BLOCK;
+		}
+		else if (!shm_mq_wait_internal(mq, &mq->mq_sender, mqh->mqh_handle))
+		{
+			mq->mq_detached = true;
+			return SHM_MQ_DETACHED;
+		}
+		mqh->mqh_counterparty_attached = true;
+	}
+
+	/* Consume any zero-copy data from previous receive operation. */
+	if (mqh->mqh_consume_pending > 0)
+	{
+		shm_mq_inc_bytes_read(mq, mqh->mqh_consume_pending);
+		mqh->mqh_consume_pending = 0;
+	}
+
+	/* Determine the message length. */
+	if (mqh->mqh_did_length_word)
+	{
+		/* We've partially received a message; recall expected length. */
+		nbytes = mqh->mqh_expected_bytes;
+	}
+	else
+	{
+		/* Try to receive the message length word. */
+		res = shm_mq_receive_bytes(mq, sizeof(uint64), nowait, &rb, &rawdata);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+		Assert(rb >= sizeof(uint64));
+		memcpy(&nbytes, rawdata, sizeof(uint64));
+		mqh->mqh_expected_bytes = nbytes;
+
+		/* If we've already got the whole message, we're done. */
+		needed = MAXALIGN64(sizeof(uint64)) + MAXALIGN64(nbytes);
+		if (rb >= needed)
+		{
+			/*
+			 * Technically, we could consume the message length information at
+			 * this point, but the extra write to shared memory wouldn't be
+			 * free and in most cases we would reap no benefit.
+			 */
+			mqh->mqh_consume_pending = needed;
+			*nbytesp = nbytes;
+			*datap = ((char *) rawdata) + MAXALIGN64(sizeof(uint64));
+			return SHM_MQ_SUCCESS;
+		}
+
+		/* Consume the length word. */
+		shm_mq_inc_bytes_read(mq, MAXALIGN64(sizeof(uint64)));
+		mqh->mqh_did_length_word = true;
+		rb -= MAXALIGN64(sizeof(uint64));
+	}
+
+	if (mqh->mqh_partial_message_bytes == 0)
+	{
+		/*
+		 * Try to obtain the whole message in a single chunk.  If this works,
+		 * we need not copy the data and can return a pointer directly into
+		 * shared memory.
+		 */
+		res = shm_mq_receive_bytes(mq, nbytes, nowait, &rb, &rawdata);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+		if (rb >= nbytes)
+		{
+			mqh->mqh_did_length_word = false;
+			mqh->mqh_consume_pending = MAXALIGN64(nbytes);
+			*nbytesp = nbytes;
+			*datap = rawdata;
+			return SHM_MQ_SUCCESS;
+		}
+
+		/*
+		 * The message has wrapped the buffer.  We'll need to copy it in order
+		 * to return it to the client in one chunk.  First, make sure we have a
+		 * large enough buffer available.
+		 */
+		if (mqh->mqh_buflen < nbytes)
+		{
+			uint64		newbuflen = Max(mqh->mqh_buflen, MQH_INITIAL_BUFSIZE);
+
+			while (newbuflen < nbytes)
+				newbuflen *= 2;
+
+			if (mqh->mqh_buffer != NULL)
+			{
+				pfree(mqh->mqh_buffer);
+				mqh->mqh_buffer = NULL;
+				mqh->mqh_buflen = 0;
+			}
+			mqh->mqh_buffer = MemoryContextAlloc(mqh->mqh_context, newbuflen);
+			mqh->mqh_buflen = newbuflen;
+		}
+	}
+
+	/* Loop until we've copied the entire message. */
+	for (;;)
+	{
+		uint64	still_needed;
+
+		/* Copy as much as we can. */
+		Assert(mqh->mqh_partial_message_bytes + rb <= nbytes);
+		memcpy(&mqh->mqh_buffer[mqh->mqh_partial_message_bytes], rawdata, rb);
+		mqh->mqh_partial_message_bytes += rb;
+
+		/*
+		 * Update count of bytes read, with alignment padding.  Note
+		 * that this will never actually insert any padding except at the
+		 * end of a message, because the buffer size is a multiple of
+		 * MAXIMUM_ALIGNOF, and each read and write is as well.
+		 */
+		Assert(mqh->mqh_partial_message_bytes == nbytes ||
+				rb == MAXALIGN64(rb));
+		shm_mq_inc_bytes_read(mq, MAXALIGN64(rb));
+
+		/* If we got all the data, exit the loop. */
+		if (mqh->mqh_partial_message_bytes >= nbytes)
+			break;
+
+		/* Wait for some more data. */
+		still_needed = nbytes - mqh->mqh_partial_message_bytes;
+		res = shm_mq_receive_bytes(mq, still_needed, nowait, &rb, &rawdata);
+		if (res != SHM_MQ_SUCCESS)
+			return res;
+		if (rb > still_needed)
+			rb = still_needed;
+	}
+
+	/* Return the complete message, and reset for next message. */
+	*nbytesp = nbytes;
+	*datap = mqh->mqh_buffer;
+	mqh->mqh_did_length_word = false;
+	mqh->mqh_partial_message_bytes = 0;
+	return SHM_MQ_SUCCESS;
+}
+
+/*
+ * Wait for the other process that's supposed to use this queue to attach
+ * to it.
+ *
+ * The return value is SHM_MQ_DETACHED if the worker has already detached or
+ * if it dies; it is SHM_MQ_SUCCESS if we detect that the worker has attached.
+ * Note that we will only be able to detect that the worker has died before
+ * attaching if a background worker handle was passed to shm_mq_attach().
+ */
+shm_mq_result
+shm_mq_wait_for_attach(shm_mq_handle *mqh)
+{
+	shm_mq	   *mq = mqh->mqh_queue;
+	PGPROC	   **victim;
+
+	if (shm_mq_get_receiver(mq) == MyProc)
+		victim = &mq->mq_sender;
+	else
+	{
+		Assert(shm_mq_get_sender(mq) == MyProc);
+		victim = &mq->mq_receiver;
+	}
+
+	if (shm_mq_wait_internal(mq, victim, mqh->mqh_handle))
+		return SHM_MQ_SUCCESS;
+	else
+		return SHM_MQ_DETACHED;
+}
+
+/*
+ * Detach a shared message queue.
+ *
+ * The purpose of this function is to make sure that the process
+ * with which we're communicating doesn't block forever waiting for us to
+ * fill or drain the queue once we've lost interest.  Whem the sender
+ * detaches, the receiver can read any messages remaining in the queue;
+ * further reads will return SHM_MQ_DETACHED.  If the receiver detaches,
+ * further attempts to send messages will likewise return SHM_MQ_DETACHED.
+ */
+void
+shm_mq_detach(shm_mq *mq)
+{
+	volatile shm_mq *vmq = mq;
+	PGPROC	   *victim;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	if (vmq->mq_sender == MyProc)
+		victim = vmq->mq_receiver;
+	else
+	{
+		Assert(vmq->mq_receiver == MyProc);
+		victim = vmq->mq_sender;
+	}
+	vmq->mq_detached = true;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (victim != NULL)
+		SetLatch(&victim->procLatch);
+}
+
+/*
+ * Write bytes into a shared message queue.
+ */
+static shm_mq_result
+shm_mq_send_bytes(shm_mq_handle *mqh, uint64 nbytes, void *data, bool nowait,
+				  uint64 *bytes_written)
+{
+	shm_mq	   *mq = mqh->mqh_queue;
+	uint64		sent = 0;
+	uint64		used;
+	uint64		ringsize = mq->mq_ring_size;
+	uint64		available;
+
+	while (sent < nbytes)
+	{
+		bool	detached;
+		uint64	rb;
+
+		/* Compute number of ring buffer bytes used and available. */
+		rb = shm_mq_get_bytes_read(mq, &detached);
+		Assert(mq->mq_bytes_written >= rb);
+		used = mq->mq_bytes_written - rb;
+		Assert(used <= ringsize);
+		available = Min(ringsize - used, nbytes - sent);
+
+		/* Bail out if the queue has been detached. */
+		if (detached)
+			return SHM_MQ_DETACHED;
+
+		if (available == 0)
+		{
+			shm_mq_result	res;
+
+			/*
+			 * The queue is full, so if the receiver isn't yet known to be
+			 * attached, we must wait for that to happen.
+			 */
+			if (!mqh->mqh_counterparty_attached)
+			{
+				if (nowait)
+				{
+					if (shm_mq_get_receiver(mq) == NULL)
+						return SHM_MQ_WOULD_BLOCK;
+				}
+				else if (!shm_mq_wait_internal(mq, &mq->mq_receiver,
+											   mqh->mqh_handle))
+				{
+					mq->mq_detached = true;
+					return SHM_MQ_DETACHED;
+				}
+				mqh->mqh_counterparty_attached = true;
+			}
+
+			/* Let the receiver know that we need them to read some data. */
+			res = shm_mq_notify_receiver(mq);
+			if (res != SHM_MQ_SUCCESS)
+			{
+				*bytes_written = res;
+				return res;
+			}
+
+			/* Skip manipulation of our latch if nowait = true. */
+			if (nowait)
+			{
+				*bytes_written = sent;
+				return SHM_MQ_WOULD_BLOCK;
+			}
+
+			/*
+			 * Wait for our latch to be set.  It might already be set for
+			 * some unrelated reason, but that'll just result in one extra
+			 * trip through the loop.  It's worth it to avoid resetting the
+			 * latch at top of loop, because setting an already-set latch is
+			 * much cheaper than setting one that has been reset.
+			 */
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+			/* An interrupt may have occurred while we were waiting. */
+			CHECK_FOR_INTERRUPTS();
+
+			/* Reset the latch so we don't spin. */
+			ResetLatch(&MyProc->procLatch);
+		}
+		else
+		{
+			uint64	offset = mq->mq_bytes_written % ringsize;
+			uint64	sendnow = Min(available, ringsize - offset);
+
+			/* Write as much data as we can via a single memcpy(). */
+			memcpy(&mq->mq_ring[mq->mq_ring_offset + offset],
+				   (char *) data + sent, sendnow);
+			sent += sendnow;
+
+			/*
+			 * Update count of bytes written, with alignment padding.  Note
+			 * that this will never actually insert any padding except at the
+			 * end of a run of bytes, because the buffer size is a multiple of
+			 * MAXIMUM_ALIGNOF, and each read is as well.
+			 */
+			Assert(sent == nbytes || sendnow == MAXALIGN64(sendnow));
+			shm_mq_inc_bytes_written(mq, MAXALIGN64(sendnow));
+
+			/*
+			 * For efficiency, we don't set the reader's latch here.  We'll
+			 * do that only when the buffer fills up or after writing an
+			 * entire message.
+			 */
+		}
+	}
+
+	*bytes_written = sent;
+	return SHM_MQ_SUCCESS;
+}
+
+/*
+ * Wait until at least *nbytesp bytes are available to be read from the
+ * shared message queue, or until the buffer wraps around.  On return,
+ * *datap is set to the location at which data bytes can be read.  The
+ * return value is the number of bytes available to be read starting at
+ * that offset; if the message has wrapped the buffer, it may be less than
+ * bytes_needed.
+ */
+static shm_mq_result
+shm_mq_receive_bytes(shm_mq *mq, uint64 bytes_needed, bool nowait,
+					 uint64 *nbytesp, void **datap)
+{
+	uint64		used;
+	uint64		ringsize = mq->mq_ring_size;
+	uint64		written;
+
+	for (;;)
+	{
+		uint64		offset;
+		bool		detached;
+
+		/* Get bytes written, so we can compute what's available to read. */
+		written = shm_mq_get_bytes_written(mq, &detached);
+		used = written - mq->mq_bytes_read;
+		Assert(used <= ringsize);
+		offset = mq->mq_bytes_read % ringsize;
+
+		/* If we have enough data or buffer has wrapped, we're done. */
+		if (used >= bytes_needed || offset + used >= ringsize)
+		{
+			*nbytesp = Min(used, ringsize - offset);
+			*datap = &mq->mq_ring[mq->mq_ring_offset + offset];
+			return SHM_MQ_SUCCESS;
+		}
+
+		/*
+		 * Fall out before waiting if the queue has been detached.
+		 *
+		 * Note that we don't check for this until *after* considering
+		 * whether the data already available is enough, since the
+		 * receiver can finish receiving a message stored in the buffer
+		 * even after the sender has detached.
+		 */
+		if (detached)
+			return SHM_MQ_DETACHED;
+
+		/* Skip manipulation of our latch if nowait = true. */
+		if (nowait)
+			return SHM_MQ_WOULD_BLOCK;
+
+		/*
+		 * Wait for our latch to be set.  It might already be set for
+		 * some unrelated reason, but that'll just result in one extra
+		 * trip through the loop.  It's worth it to avoid resetting the
+		 * latch at top of loop, because setting an already-set latch is
+		 * much cheaper than setting one that has been reset.
+		 */
+		WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+		/* An interrupt may have occurred while we were waiting. */
+		CHECK_FOR_INTERRUPTS();
+
+		/* Reset the latch so we don't spin. */
+		ResetLatch(&MyProc->procLatch);
+	}
+}
+
+/*
+ * This is used when a process is waiting for its counterpart to attach to the
+ * queue.  We exit when the other process attaches as expected, or, if
+ * handle != NULL, when the referenced background process or the postmaster
+ * dies.  Note that if handle == NULL, and the process fails to attach, we'll
+ * potentially get stuck here forever waiting for a process that may never
+ * start.  We do check for interrupts, though.
+ *
+ * ptr is a pointer to the memory address that we're expecting to become
+ * non-NULL when our counterpart attaches to the queue.
+ */
+static bool
+shm_mq_wait_internal(volatile shm_mq *mq, PGPROC * volatile *ptr,
+					 BackgroundWorkerHandle *handle)
+{
+	bool	save_set_latch_on_sigusr1;
+	bool	result = false;
+
+	save_set_latch_on_sigusr1 = set_latch_on_sigusr1;
+	if (handle != NULL)
+		set_latch_on_sigusr1 = true;
+
+	PG_TRY();
+	{
+		for (;;)
+		{
+			BgwHandleStatus	status;
+			pid_t	pid;
+			bool	detached;
+
+			/* Acquire the lock just long enough to check the pointer. */
+			SpinLockAcquire(&mq->mq_mutex);
+			detached = mq->mq_detached;
+			result = (*ptr != NULL);
+			SpinLockRelease(&mq->mq_mutex);
+
+			/* Fail if detached; else succeed if initialized. */
+			if (detached)
+			{
+				result = false;
+				break;
+			}
+			if (result)
+				break;
+
+			if (handle != NULL)
+			{
+				/* Check for unexpected worker death. */
+				status = GetBackgroundWorkerPid(handle, &pid);
+				if (status != BGWH_STARTED && status != BGWH_NOT_YET_STARTED)
+				{
+					result = false;
+					break;
+				}
+			}
+
+			/* Wait to be signalled. */
+			WaitLatch(&MyProc->procLatch, WL_LATCH_SET, 0);
+
+			/* An interrupt may have occurred while we were waiting. */
+			CHECK_FOR_INTERRUPTS();
+
+			/* Reset the latch so we don't spin. */
+			ResetLatch(&MyProc->procLatch);
+		}
+	}
+	PG_CATCH();
+	{
+		set_latch_on_sigusr1 = save_set_latch_on_sigusr1;
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
+
+	return result;
+}
+
+/*
+ * Get the number of bytes read.  The receiver need not use this to access
+ * the count of bytes read, but the sender must.
+ */
+static uint64
+shm_mq_get_bytes_read(volatile shm_mq *mq, bool *detached)
+{
+	uint64	v;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	v = mq->mq_bytes_read;
+	*detached = mq->mq_detached;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return v;
+}
+
+/*
+ * Increment the number of bytes read.
+ */
+static void
+shm_mq_inc_bytes_read(volatile shm_mq *mq, uint64 n)
+{
+	PGPROC	   *sender;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	mq->mq_bytes_read += n;
+	sender = mq->mq_sender;
+	SpinLockRelease(&mq->mq_mutex);
+
+	/* We shoudn't have any bytes to read without a sender. */
+	Assert(sender != NULL);
+	SetLatch(&sender->procLatch);
+}
+
+/*
+ * Get the number of bytes written.  The sender need not use this to access
+ * the count of bytes written, but the reciever must.
+ */
+static uint64
+shm_mq_get_bytes_written(volatile shm_mq *mq, bool *detached)
+{
+	uint64	v;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	v = mq->mq_bytes_written;
+	*detached = mq->mq_detached;
+	SpinLockRelease(&mq->mq_mutex);
+
+	return v;
+}
+
+/*
+ * Increment the number of bytes written.
+ */
+static void
+shm_mq_inc_bytes_written(volatile shm_mq *mq, uint64 n)
+{
+	SpinLockAcquire(&mq->mq_mutex);
+	mq->mq_bytes_written += n;
+	SpinLockRelease(&mq->mq_mutex);
+}
+
+/*
+ * Set sender's latch, unless queue is detached.
+ */
+static shm_mq_result
+shm_mq_notify_receiver(volatile shm_mq *mq)
+{
+	PGPROC *receiver;
+	bool	detached;
+
+	SpinLockAcquire(&mq->mq_mutex);
+	detached = mq->mq_detached;
+	receiver = mq->mq_receiver;
+	SpinLockRelease(&mq->mq_mutex);
+
+	if (detached)
+		return SHM_MQ_DETACHED;
+	if (receiver)
+		SetLatch(&receiver->procLatch);
+	return SHM_MQ_SUCCESS;
+}
+
+/* Shim for on_dsm_callback. */
+static void
+shm_mq_detach_callback(dsm_segment *seg, Datum arg)
+{
+	shm_mq	   *mq = (shm_mq *) DatumGetPointer(arg);
+
+	shm_mq_detach(mq);
+}
diff --git a/src/include/storage/shm_mq.h b/src/include/storage/shm_mq.h
new file mode 100644
index 0000000..1ce88a1
--- /dev/null
+++ b/src/include/storage/shm_mq.h
@@ -0,0 +1,70 @@
+/*-------------------------------------------------------------------------
+ *
+ * shm_mq.h
+ *	  single-reader, single-writer shared memory message queue
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/storage/shm_mq.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SHM_MQ_H
+#define SHM_MQ_H
+
+#include "postmaster/bgworker.h"
+#include "storage/dsm.h"
+#include "storage/proc.h"
+
+/* The queue itself, in shared memory. */
+struct shm_mq;
+typedef struct shm_mq shm_mq;
+
+/* Backend-private state. */
+struct shm_mq_handle;
+typedef struct shm_mq_handle shm_mq_handle;
+
+/* Possible results of a send or receive operation. */
+typedef enum
+{
+	SHM_MQ_SUCCESS,			/* Sent or received a message. */
+	SHM_MQ_WOULD_BLOCK,		/* Not completed; retry later. */
+	SHM_MQ_DETACHED			/* Other process has detached queue. */
+} shm_mq_result;
+
+/*
+ * Primitives to create a queue and set the sender and receiver.
+ *
+ * Both the sender and the receiver must be set before any messages are read
+ * or written, but they need not be set by the same process.  Each must be
+ * set exactly once.
+ */
+extern shm_mq *shm_mq_create(void *address, Size size);
+extern void shm_mq_set_receiver(shm_mq *mq, PGPROC *);
+extern void shm_mq_set_sender(shm_mq *mq, PGPROC *);
+
+/* Accessor methods for sender and receiver. */
+extern PGPROC *shm_mq_get_receiver(shm_mq *);
+extern PGPROC *shm_mq_get_sender(shm_mq *);
+
+/* Set up backend-local queue state. */
+extern shm_mq_handle *shm_mq_attach(shm_mq *mq, dsm_segment *seg,
+			  BackgroundWorkerHandle *handle);
+
+/* Break connection. */
+extern void shm_mq_detach(shm_mq *);
+
+/* Send or receive messages. */
+extern shm_mq_result shm_mq_send(shm_mq_handle *mqh,
+			uint64 nbytes, void *data, bool nowait);
+extern shm_mq_result shm_mq_receive(shm_mq_handle *mqh,
+			   uint64 *nbytesp, void **datap, bool nowait);
+
+/* Wait for our counterparty to attach to the queue. */
+extern shm_mq_result shm_mq_wait_for_attach(shm_mq_handle *mqh);
+
+/* Smallest possible queue. */
+extern const Size shm_mq_minimum_size;
+
+#endif   /* SHM_MQ_H */