Extend the ProcSignal mechanism to support barriers.

A new function EmitProcSignalBarrier() can be used to emit a global
barrier which all backends that participate in the ProcSignal
mechanism must absorb, and a new function WaitForProcSignalBarrier()
can be used to wait until all relevant backends have in fact
absorbed the barrier.

This can be used to coordinate global state changes, such as turning
checksums on while the system is running.

There's no real client of this mechanism yet, although two are
proposed, but an enum has to have at least one element, so this
includes a placeholder type (PROCSIGNAL_BARRIER_PLACEHOLDER) which
should be replaced by the first real client of this mechanism to
get committed.

Andres Freund and Robert Haas, reviewed by Daniel Gustafsson and,
in earlier versions, by Magnus Hagander.

Discussion: http://postgr.es/m/CA+TgmoZwDk=BguVDVa+qdA6SBKef=PKbaKDQALTC_9qoz1mJqg@mail.gmail.com

1 files changed, 290 insertions, 11 deletions

diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index fde97a1036b..06e00eae15b 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -20,6 +20,7 @@
 #include "access/parallel.h"
 #include "commands/async.h"
 #include "miscadmin.h"
+#include "pgstat.h"
 #include "replication/walsender.h"
 #include "storage/ipc.h"
 #include "storage/latch.h"
@@ -45,25 +46,53 @@
  * The flags are actually declared as "volatile sig_atomic_t" for maximum
  * portability.  This should ensure that loads and stores of the flag
  * values are atomic, allowing us to dispense with any explicit locking.
+ *
+ * pss_signalFlags are intended to be set in cases where we don't need to
+ * keep track of whether or not the target process has handled the signal,
+ * but sometimes we need confirmation, as when making a global state change
+ * that cannot be considered complete until all backends have taken notice
+ * of it. For such use cases, we set a bit in pss_barrierCheckMask and then
+ * increment the current "barrier generation"; when the new barrier generation
+ * (or greater) appears in the pss_barrierGeneration flag of every process,
+ * we know that the message has been received everywhere.
  */
 typedef struct
 {
 	pid_t		pss_pid;
 	sig_atomic_t pss_signalFlags[NUM_PROCSIGNALS];
+	pg_atomic_uint64	pss_barrierGeneration;
+	pg_atomic_uint32	pss_barrierCheckMask;
 } ProcSignalSlot;
 
 /*
+ * Information that is global to the entire ProcSignal system can be stored
+ * here.
+ *
+ * psh_barrierGeneration is the highest barrier generation in existence.
+ */
+typedef struct
+{
+	pg_atomic_uint64	psh_barrierGeneration;
+	ProcSignalSlot		psh_slot[FLEXIBLE_ARRAY_MEMBER];
+} ProcSignalHeader;
+
+/*
  * We reserve a slot for each possible BackendId, plus one for each
  * possible auxiliary process type.  (This scheme assumes there is not
  * more than one of any auxiliary process type at a time.)
  */
 #define NumProcSignalSlots	(MaxBackends + NUM_AUXPROCTYPES)
 
-static ProcSignalSlot *ProcSignalSlots = NULL;
+/* Check whether the relevant type bit is set in the flags. */
+#define BARRIER_SHOULD_CHECK(flags, type) \
+	(((flags) & (((uint32) 1) << (uint32) (type))) != 0)
+
+static ProcSignalHeader *ProcSignal = NULL;
 static volatile ProcSignalSlot *MyProcSignalSlot = NULL;
 
 static bool CheckProcSignal(ProcSignalReason reason);
 static void CleanupProcSignalState(int status, Datum arg);
+static void ProcessBarrierPlaceholder(void);
 
 /*
  * ProcSignalShmemSize
@@ -72,7 +101,11 @@ static void CleanupProcSignalState(int status, Datum arg);
 Size
 ProcSignalShmemSize(void)
 {
-	return NumProcSignalSlots * sizeof(ProcSignalSlot);
+	Size	size;
+
+	size = mul_size(NumProcSignalSlots, sizeof(ProcSignalSlot));
+	size = add_size(size, offsetof(ProcSignalHeader, psh_slot));
+	return size;
 }
 
 /*
@@ -85,12 +118,26 @@ ProcSignalShmemInit(void)
 	Size		size = ProcSignalShmemSize();
 	bool		found;
 
-	ProcSignalSlots = (ProcSignalSlot *)
-		ShmemInitStruct("ProcSignalSlots", size, &found);
+	ProcSignal = (ProcSignalHeader *)
+		ShmemInitStruct("ProcSignal", size, &found);
 
-	/* If we're first, set everything to zeroes */
+	/* If we're first, initialize. */
 	if (!found)
-		MemSet(ProcSignalSlots, 0, size);
+	{
+		int		i;
+
+		pg_atomic_init_u64(&ProcSignal->psh_barrierGeneration, 0);
+
+		for (i = 0; i < NumProcSignalSlots; ++i)
+		{
+			ProcSignalSlot *slot = &ProcSignal->psh_slot[i];
+
+			slot->pss_pid = 0;
+			MemSet(slot->pss_signalFlags, 0, sizeof(slot->pss_signalFlags));
+			pg_atomic_init_u64(&slot->pss_barrierGeneration, PG_UINT64_MAX);
+			pg_atomic_init_u32(&slot->pss_barrierCheckMask, 0);
+		}
+	}
 }
 
 /*
@@ -104,10 +151,11 @@ void
 ProcSignalInit(int pss_idx)
 {
 	volatile ProcSignalSlot *slot;
+	uint64		barrier_generation;
 
 	Assert(pss_idx >= 1 && pss_idx <= NumProcSignalSlots);
 
-	slot = &ProcSignalSlots[pss_idx - 1];
+	slot = &ProcSignal->psh_slot[pss_idx - 1];
 
 	/* sanity check */
 	if (slot->pss_pid != 0)
@@ -117,6 +165,23 @@ ProcSignalInit(int pss_idx)
 	/* Clear out any leftover signal reasons */
 	MemSet(slot->pss_signalFlags, 0, NUM_PROCSIGNALS * sizeof(sig_atomic_t));
 
+	/*
+	 * Initialize barrier state. Since we're a brand-new process, there
+	 * shouldn't be any leftover backend-private state that needs to be
+	 * updated. Therefore, we can broadcast the latest barrier generation
+	 * and disregard any previously-set check bits.
+	 *
+	 * NB: This only works if this initialization happens early enough in the
+	 * startup sequence that we haven't yet cached any state that might need
+	 * to be invalidated. That's also why we have a memory barrier here, to
+	 * be sure that any later reads of memory happen strictly after this.
+	 */
+	pg_atomic_write_u32(&slot->pss_barrierCheckMask, 0);
+	barrier_generation =
+		pg_atomic_read_u64(&ProcSignal->psh_barrierGeneration);
+	pg_atomic_write_u64(&slot->pss_barrierGeneration, barrier_generation);
+	pg_memory_barrier();
+
 	/* Mark slot with my PID */
 	slot->pss_pid = MyProcPid;
 
@@ -129,7 +194,7 @@ ProcSignalInit(int pss_idx)
 
 /*
  * CleanupProcSignalState
- *		Remove current process from ProcSignalSlots
+ *		Remove current process from ProcSignal mechanism
  *
  * This function is called via on_shmem_exit() during backend shutdown.
  */
@@ -139,7 +204,7 @@ CleanupProcSignalState(int status, Datum arg)
 	int			pss_idx = DatumGetInt32(arg);
 	volatile ProcSignalSlot *slot;
 
-	slot = &ProcSignalSlots[pss_idx - 1];
+	slot = &ProcSignal->psh_slot[pss_idx - 1];
 	Assert(slot == MyProcSignalSlot);
 
 	/*
@@ -161,6 +226,12 @@ CleanupProcSignalState(int status, Datum arg)
 		return;					/* XXX better to zero the slot anyway? */
 	}
 
+	/*
+	 * Make this slot look like it's absorbed all possible barriers, so that
+	 * no barrier waits block on it.
+	 */
+	pg_atomic_write_u64(&slot->pss_barrierGeneration, PG_UINT64_MAX);
+
 	slot->pss_pid = 0;
 }
 
@@ -182,7 +253,7 @@ SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
 
 	if (backendId != InvalidBackendId)
 	{
-		slot = &ProcSignalSlots[backendId - 1];
+		slot = &ProcSignal->psh_slot[backendId - 1];
 
 		/*
 		 * Note: Since there's no locking, it's possible that the target
@@ -212,7 +283,7 @@ SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
 
 		for (i = NumProcSignalSlots - 1; i >= 0; i--)
 		{
-			slot = &ProcSignalSlots[i];
+			slot = &ProcSignal->psh_slot[i];
 
 			if (slot->pss_pid == pid)
 			{
@@ -231,6 +302,187 @@ SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
 }
 
 /*
+ * EmitProcSignalBarrier
+ *		Send a signal to every Postgres process
+ *
+ * The return value of this function is the barrier "generation" created
+ * by this operation. This value can be passed to WaitForProcSignalBarrier
+ * to wait until it is known that every participant in the ProcSignal
+ * mechanism has absorbed the signal (or started afterwards).
+ *
+ * Note that it would be a bad idea to use this for anything that happens
+ * frequently, as interrupting every backend could cause a noticeable
+ * performance hit.
+ *
+ * Callers are entitled to assume that this function will not throw ERROR
+ * or FATAL.
+ */
+uint64
+EmitProcSignalBarrier(ProcSignalBarrierType type)
+{
+	uint64	flagbit = UINT64CONST(1) << (uint64) type;
+	uint64	generation;
+
+	/*
+	 * Set all the flags.
+	 *
+	 * Note that pg_atomic_fetch_or_u32 has full barrier semantics, so this
+	 * is totally ordered with respect to anything the caller did before, and
+	 * anything that we do afterwards. (This is also true of the later call
+	 * to pg_atomic_add_fetch_u64.)
+	 */
+	for (int i = 0; i < NumProcSignalSlots; i++)
+	{
+		volatile ProcSignalSlot *slot = &ProcSignal->psh_slot[i];
+
+		pg_atomic_fetch_or_u32(&slot->pss_barrierCheckMask, flagbit);
+	}
+
+	/*
+	 * Increment the generation counter.
+	 */
+	generation =
+		pg_atomic_add_fetch_u64(&ProcSignal->psh_barrierGeneration, 1);
+
+	/*
+	 * Signal all the processes, so that they update their advertised barrier
+	 * generation.
+	 *
+	 * Concurrency is not a problem here. Backends that have exited don't
+	 * matter, and new backends that have joined since we entered this function
+	 * must already have current state, since the caller is responsible for
+	 * making sure that the relevant state is entirely visible before calling
+	 * this function in the first place. We still have to wake them up -
+	 * because we can't distinguish between such backends and older backends
+	 * that need to update state - but they won't actually need to change
+	 * any state.
+	 */
+	for (int i = NumProcSignalSlots - 1; i >= 0; i--)
+	{
+		volatile ProcSignalSlot *slot = &ProcSignal->psh_slot[i];
+		pid_t	pid = slot->pss_pid;
+
+		if (pid != 0)
+			kill(pid, SIGUSR1);
+	}
+
+	return generation;
+}
+
+/*
+ * WaitForProcSignalBarrier - wait until it is guaranteed that all changes
+ * requested by a specific call to EmitProcSignalBarrier() have taken effect.
+ *
+ * We expect that the barrier will normally be absorbed very quickly by other
+ * backends, so we start by waiting just 1/8 of a second and then back off
+ * by a factor of two every time we time out, to a maximum wait time of
+ * 1 second.
+ */
+void
+WaitForProcSignalBarrier(uint64 generation)
+{
+	long	timeout = 125L;
+
+	for (int i = NumProcSignalSlots - 1; i >= 0; i--)
+	{
+		volatile ProcSignalSlot *slot = &ProcSignal->psh_slot[i];
+		uint64	oldval;
+
+		oldval = pg_atomic_read_u64(&slot->pss_barrierGeneration);
+		while (oldval < generation)
+		{
+			int		events;
+
+			CHECK_FOR_INTERRUPTS();
+
+			events =
+				WaitLatch(MyLatch,
+						  WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+						  timeout, WAIT_EVENT_PROC_SIGNAL_BARRIER);
+			ResetLatch(MyLatch);
+
+			oldval = pg_atomic_read_u64(&slot->pss_barrierGeneration);
+			if (events & WL_TIMEOUT)
+				timeout = Min(timeout * 2, 1000L);
+		}
+	}
+
+	/*
+	 * The caller is probably calling this function because it wants to
+	 * read the shared state or perform further writes to shared state once
+	 * all backends are known to have absorbed the barrier. However, the
+	 * read of pss_barrierGeneration was performed unlocked; insert a memory
+	 * barrier to separate it from whatever follows.
+	 */
+	pg_memory_barrier();
+}
+
+/*
+ * Perform global barrier related interrupt checking.
+ *
+ * Any backend that participates in ProcSignal signalling must arrange to
+ * call this function periodically. It is called from CHECK_FOR_INTERRUPTS(),
+ * which is enough for normal backends, but not necessarily for all types of
+ * background processes.
+ */
+void
+ProcessProcSignalBarrier(void)
+{
+	uint64 generation;
+	uint32 flags;
+
+	/* Exit quickly if there's no work to do. */
+	if (!ProcSignalBarrierPending)
+		return;
+	ProcSignalBarrierPending = false;
+
+	/*
+	 * Read the current barrier generation, and then get the flags that
+	 * are set for this backend. Note that pg_atomic_exchange_u32 is a full
+	 * barrier, so we're guaranteed that the read of the barrier generation
+	 * happens before we atomically extract the flags, and that any subsequent
+	 * state changes happen afterward.
+	 */
+	generation = pg_atomic_read_u64(&ProcSignal->psh_barrierGeneration);
+	flags = pg_atomic_exchange_u32(&MyProcSignalSlot->pss_barrierCheckMask, 0);
+
+	/*
+	 * Process each type of barrier. It's important that nothing we call from
+	 * here throws an error, because pss_barrierCheckMask has already been
+	 * cleared. If we jumped out of here before processing all barrier types,
+	 * then we'd forget about the need to do so later.
+	 *
+	 * NB: It ought to be OK to call the barrier-processing functions
+	 * unconditionally, but it's more efficient to call only the ones that
+	 * might need us to do something based on the flags.
+	 */
+	if (BARRIER_SHOULD_CHECK(flags, PROCSIGNAL_BARRIER_PLACEHOLDER))
+		ProcessBarrierPlaceholder();
+
+	/*
+	 * State changes related to all types of barriers that might have been
+	 * emitted have now been handled, so we can update our notion of the
+	 * generation to the one we observed before beginning the updates. If
+	 * things have changed further, it'll get fixed up when this function is
+	 * next called.
+	 */
+	pg_atomic_write_u64(&MyProcSignalSlot->pss_barrierGeneration, generation);
+}
+
+static void
+ProcessBarrierPlaceholder(void)
+{
+	/*
+	 * XXX. This is just a placeholder until the first real user of this
+	 * machinery gets committed. Rename PROCSIGNAL_BARRIER_PLACEHOLDER to
+	 * PROCSIGNAL_BARRIER_SOMETHING_ELSE where SOMETHING_ELSE is something
+	 * appropriately descriptive. Get rid of this function and instead have
+	 * ProcessBarrierSomethingElse. Most likely, that function should live
+	 * in the file pertaining to that subsystem, rather than here.
+	 */
+}
+
+/*
  * CheckProcSignal - check to see if a particular reason has been
  * signaled, and clear the signal flag.  Should be called after receiving
  * SIGUSR1.
@@ -254,6 +506,27 @@ CheckProcSignal(ProcSignalReason reason)
 }
 
 /*
+ * CheckProcSignalBarrier - check for new barriers we need to absorb
+ */
+static bool
+CheckProcSignalBarrier(void)
+{
+	volatile ProcSignalSlot *slot = MyProcSignalSlot;
+
+	if (slot != NULL)
+	{
+		uint64	mygen;
+		uint64	curgen;
+
+		mygen = pg_atomic_read_u64(&slot->pss_barrierGeneration);
+		curgen = pg_atomic_read_u64(&ProcSignal->psh_barrierGeneration);
+		return (mygen != curgen);
+	}
+
+	return false;
+}
+
+/*
  * procsignal_sigusr1_handler - handle SIGUSR1 signal.
  */
 void
@@ -291,6 +564,12 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
 	if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN))
 		RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
 
+	if (CheckProcSignalBarrier())
+	{
+		InterruptPending = true;
+		ProcSignalBarrierPending = true;
+	}
+
 	SetLatch(MyLatch);
 
 	latch_sigusr1_handler();