Replace the pg_listener-based LISTEN/NOTIFY mechanism with an in-memory queue.

In addition, add support for a "payload" string to be passed along with
each notify event.

This implementation should be significantly more efficient than the old one,
and is also more compatible with Hot Standby usage.  There is not yet any
facility for HS slaves to receive notifications generated on the master,
although such a thing is possible in future.

Joachim Wieland, reviewed by Jeff Davis; also hacked on by me.

8 files changed, 59 insertions, 80 deletions

diff --git a/src/include/access/slru.h b/src/include/access/slru.h
index 8e820ae72dd..4cc40ba5f70 100644
--- a/src/include/access/slru.h
+++ b/src/include/access/slru.h
@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/access/slru.h,v 1.25 2010/01/02 16:58:00 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/access/slru.h,v 1.26 2010/02/16 22:34:50 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -18,6 +18,25 @@
 
 
 /*
+ * Define SLRU segment size.  A page is the same BLCKSZ as is used everywhere
+ * else in Postgres.  The segment size can be chosen somewhat arbitrarily;
+ * we make it 32 pages by default, or 256Kb, i.e. 1M transactions for CLOG
+ * or 64K transactions for SUBTRANS.
+ *
+ * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF,
+ * page numbering also wraps around at 0xFFFFFFFF/xxxx_XACTS_PER_PAGE (where
+ * xxxx is CLOG or SUBTRANS, respectively), and segment numbering at
+ * 0xFFFFFFFF/xxxx_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT.  We need
+ * take no explicit notice of that fact in slru.c, except when comparing
+ * segment and page numbers in SimpleLruTruncate (see PagePrecedes()).
+ *
+ * Note: slru.c currently assumes that segment file names will be four hex
+ * digits.  This sets a lower bound on the segment size (64K transactions
+ * for 32-bit TransactionIds).
+ */
+#define SLRU_PAGES_PER_SEGMENT	32
+
+/*
  * Page status codes.  Note that these do not include the "dirty" bit.
  * page_dirty can be TRUE only in the VALID or WRITE_IN_PROGRESS states;
  * in the latter case it implies that the page has been re-dirtied since
@@ -55,8 +74,8 @@ typedef struct SlruSharedData
 	/*
 	 * Optional array of WAL flush LSNs associated with entries in the SLRU
 	 * pages.  If not zero/NULL, we must flush WAL before writing pages (true
-	 * for pg_clog, false for multixact and pg_subtrans).  group_lsn[] has
-	 * lsn_groups_per_page entries per buffer slot, each containing the
+	 * for pg_clog, false for multixact, pg_subtrans, pg_notify).  group_lsn[]
+	 * has lsn_groups_per_page entries per buffer slot, each containing the
 	 * highest LSN known for a contiguous group of SLRU entries on that slot's
 	 * page.
 	 */
@@ -94,7 +113,7 @@ typedef struct SlruCtlData
 
 	/*
 	 * This flag tells whether to fsync writes (true for pg_clog and multixact
-	 * stuff, false for pg_subtrans).
+	 * stuff, false for pg_subtrans and pg_notify).
 	 */
 	bool		do_fsync;
 
diff --git a/src/include/access/twophase_rmgr.h b/src/include/access/twophase_rmgr.h
index a42d7745205..1d4d1cb2217 100644
--- a/src/include/access/twophase_rmgr.h
+++ b/src/include/access/twophase_rmgr.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/access/twophase_rmgr.h,v 1.11 2010/01/02 16:58:00 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/access/twophase_rmgr.h,v 1.12 2010/02/16 22:34:50 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -23,9 +23,8 @@ typedef uint8 TwoPhaseRmgrId;
  */
 #define TWOPHASE_RM_END_ID			0
 #define TWOPHASE_RM_LOCK_ID			1
-#define TWOPHASE_RM_NOTIFY_ID		2
-#define TWOPHASE_RM_PGSTAT_ID		3
-#define TWOPHASE_RM_MULTIXACT_ID	4
+#define TWOPHASE_RM_PGSTAT_ID		2
+#define TWOPHASE_RM_MULTIXACT_ID	3
 #define TWOPHASE_RM_MAX_ID			TWOPHASE_RM_MULTIXACT_ID
 
 extern const TwoPhaseCallback twophase_recover_callbacks[];
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 9a1d19380c7..2d63232b9b5 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -37,7 +37,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.584 2010/02/12 17:33:20 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.585 2010/02/16 22:34:54 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201002121
+#define CATALOG_VERSION_NO	201002161
 
 #endif
diff --git a/src/include/catalog/pg_listener.h b/src/include/catalog/pg_listener.h
deleted file mode 100644
index 4d1a77d6602..00000000000
--- a/src/include/catalog/pg_listener.h
+++ /dev/null
@@ -1,59 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * pg_listener.h
- *	  Asynchronous notification
- *
- *
- * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- * $PostgreSQL: pgsql/src/include/catalog/pg_listener.h,v 1.28 2010/01/05 01:06:56 tgl Exp $
- *
- * NOTES
- *	  the genbki.pl script reads this file and generates .bki
- *	  information from the DATA() statements.
- *
- *-------------------------------------------------------------------------
- */
-#ifndef PG_LISTENER_H
-#define PG_LISTENER_H
-
-#include "catalog/genbki.h"
-
-/* ----------------------------------------------------------------
- *		pg_listener definition.
- *
- *		cpp turns this into typedef struct FormData_pg_listener
- * ----------------------------------------------------------------
- */
-#define ListenerRelationId	2614
-
-CATALOG(pg_listener,2614) BKI_WITHOUT_OIDS
-{
-	NameData	relname;
-	int4		listenerpid;
-	int4		notification;
-} FormData_pg_listener;
-
-/* ----------------
- *		Form_pg_listener corresponds to a pointer to a tuple with
- *		the format of pg_listener relation.
- * ----------------
- */
-typedef FormData_pg_listener *Form_pg_listener;
-
-/* ----------------
- *		compiler constants for pg_listener
- * ----------------
- */
-#define Natts_pg_listener						3
-#define Anum_pg_listener_relname				1
-#define Anum_pg_listener_listenerpid			2
-#define Anum_pg_listener_notification			3
-
-/* ----------------
- *		initial contents of pg_listener are NOTHING.
- * ----------------
- */
-
-#endif   /* PG_LISTENER_H */
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 727b13e264c..1c87a94a0de 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_proc.h,v 1.568 2010/02/07 20:48:11 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_proc.h,v 1.569 2010/02/16 22:34:56 tgl Exp $
  *
  * NOTES
  *	  The script catalog/genbki.pl reads this file and generates .bki
@@ -4131,8 +4131,12 @@ DATA(insert OID = 2599 (  pg_timezone_abbrevs	PGNSP PGUID 12 1 1000 0 f f f t t
 DESCR("get the available time zone abbreviations");
 DATA(insert OID = 2856 (  pg_timezone_names		PGNSP PGUID 12 1 1000 0 f f f t t s 0 0 2249 "" "{25,25,1186,16}" "{o,o,o,o}" "{name,abbrev,utc_offset,is_dst}" _null_ pg_timezone_names _null_ _null_ _null_ ));
 DESCR("get the available time zone names");
-DATA(insert OID = 2730 (  pg_get_triggerdef    PGNSP PGUID 12 1 0 0 f f f t f s 2 0 25 "26 16" _null_ _null_ _null_ _null_ pg_get_triggerdef_ext _null_ _null_ _null_ ));
+DATA(insert OID = 2730 (  pg_get_triggerdef		PGNSP PGUID 12 1 0 0 f f f t f s 2 0 25 "26 16" _null_ _null_ _null_ _null_ pg_get_triggerdef_ext _null_ _null_ _null_ ));
 DESCR("trigger description with pretty-print option");
+DATA(insert OID = 3035 (  pg_listening_channels	PGNSP PGUID 12 1 10 0 f f f t t s 0 0 25 "" _null_ _null_ _null_ _null_ pg_listening_channels _null_ _null_ _null_ ));
+DESCR("get the channels that the current backend listens to");
+DATA(insert OID = 3036 (  pg_notify				PGNSP PGUID 12 1 0 0 f f f f f v 2 0 2278 "25 25" _null_ _null_ _null_ _null_ pg_notify _null_ _null_ _null_ ));
+DESCR("send a notification event");
 
 /* non-persistent series generator */
 DATA(insert OID = 1066 (  generate_series PGNSP PGUID 12 1 1000 0 f f f t t i 3 0 23 "23 23 23" _null_ _null_ _null_ _null_ generate_series_step_int4 _null_ _null_ _null_ ));
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index 5c9e8ab8906..a9e4d42853d 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -6,28 +6,44 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/commands/async.h,v 1.39 2010/01/02 16:58:03 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/commands/async.h,v 1.40 2010/02/16 22:34:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 #ifndef ASYNC_H
 #define ASYNC_H
 
+#include "fmgr.h"
+
+/*
+ * The number of SLRU page buffers we use for the notification queue.
+ */
+#define NUM_ASYNC_BUFFERS	8
+
 extern bool Trace_notify;
 
+extern Size AsyncShmemSize(void);
+extern void AsyncShmemInit(void);
+
 /* notify-related SQL statements */
-extern void Async_Notify(const char *relname);
-extern void Async_Listen(const char *relname);
-extern void Async_Unlisten(const char *relname);
+extern void Async_Notify(const char *channel, const char *payload);
+extern void Async_Listen(const char *channel);
+extern void Async_Unlisten(const char *channel);
 extern void Async_UnlistenAll(void);
 
+/* notify-related SQL functions */
+extern Datum pg_listening_channels(PG_FUNCTION_ARGS);
+extern Datum pg_notify(PG_FUNCTION_ARGS);
+
 /* perform (or cancel) outbound notify processing at transaction commit */
+extern void PreCommit_Notify(void);
 extern void AtCommit_Notify(void);
 extern void AtAbort_Notify(void);
 extern void AtSubStart_Notify(void);
 extern void AtSubCommit_Notify(void);
 extern void AtSubAbort_Notify(void);
 extern void AtPrepare_Notify(void);
+extern void ProcessCompletedNotifies(void);
 
 /* signal handler for inbound notifies (PROCSIG_NOTIFY_INTERRUPT) */
 extern void HandleNotifyInterrupt(void);
@@ -40,7 +56,4 @@ extern void HandleNotifyInterrupt(void);
 extern void EnableNotifyInterrupt(void);
 extern bool DisableNotifyInterrupt(void);
 
-extern void notify_twophase_postcommit(TransactionId xid, uint16 info,
-						   void *recdata, uint32 len);
-
 #endif   /* ASYNC_H */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 0c3aecfa6e6..ca229c8e23d 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -13,7 +13,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/nodes/parsenodes.h,v 1.429 2010/02/12 17:33:20 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/nodes/parsenodes.h,v 1.430 2010/02/16 22:34:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2097,6 +2097,7 @@ typedef struct NotifyStmt
 {
 	NodeTag		type;
 	char	   *conditionname;	/* condition name to notify */
+	char	   *payload;		/* the payload string, or NULL if none */
 } NotifyStmt;
 
 /* ----------------------
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index f0beb20a24b..2ace9585009 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/storage/lwlock.h,v 1.44 2010/02/07 20:48:13 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/storage/lwlock.h,v 1.45 2010/02/16 22:34:57 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -68,6 +68,8 @@ typedef enum LWLockId
 	AutovacuumScheduleLock,
 	SyncScanLock,
 	RelationMappingLock,
+ 	AsyncCtlLock,
+ 	AsyncQueueLock,
 	/* Individual lock IDs end here */
 	FirstBufMappingLock,
 	FirstLockMgrLock = FirstBufMappingLock + NUM_BUFFER_PARTITIONS,