latches-1.patch
text/x-diff
Filename: latches-1.patch
Type: text/x-diff
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| src/backend/access/transam/twophase.c | 22 | 0 |
| src/backend/access/transam/xact.c | 8 | 0 |
| src/backend/replication/walsender.c | 72 | 28 |
| src/backend/storage/ipc/ipci.c | 3 | 0 |
| src/backend/storage/ipc/latch.c | 306 | 0 |
| src/backend/storage/ipc/Makefile | 2 | 2 |
| src/backend/storage/ipc/procsignal.c | 2 | 0 |
| src/include/replication/walsender.h | 16 | 0 |
| src/include/storage/latch.h | 38 | 0 |
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 615a7fa..094d0c9 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -55,6 +55,7 @@
#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
+#include "replication/walsender.h"
#include "storage/fd.h"
#include "storage/procarray.h"
#include "storage/sinvaladt.h"
@@ -1025,6 +1026,13 @@ EndPrepare(GlobalTransaction gxact)
/* If we crash now, we have prepared: WAL replay will fix things */
+ /*
+ * Wake up all walsenders to send WAL up to the PREPARE record
+ * immediately if replication is enabled
+ */
+ if (max_wal_senders > 0)
+ WalSndWakeup();
+
/* write correct CRC and close file */
if ((write(fd, &statefile_crc, sizeof(pg_crc32))) != sizeof(pg_crc32))
{
@@ -2005,6 +2013,13 @@ RecordTransactionCommitPrepared(TransactionId xid,
/* Flush XLOG to disk */
XLogFlush(recptr);
+ /*
+ * Wake up all walsenders to send WAL up to the COMMIT PREPARED record
+ * immediately if replication is enabled
+ */
+ if (max_wal_senders > 0)
+ WalSndWakeup();
+
/* Mark the transaction committed in pg_clog */
TransactionIdCommitTree(xid, nchildren, children);
@@ -2078,6 +2093,13 @@ RecordTransactionAbortPrepared(TransactionId xid,
XLogFlush(recptr);
/*
+ * Wake up all walsenders to send WAL up to the ABORT PREPARED record
+ * immediately if replication is enabled
+ */
+ if (max_wal_senders > 0)
+ WalSndWakeup();
+
+ /*
* Mark the transaction aborted in clog. This is not absolutely necessary
* but we may as well do it while we are here.
*/
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 6bcc55c..942d5c2 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -36,6 +36,7 @@
#include "libpq/be-fsstubs.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "replication/walsender.h"
#include "storage/bufmgr.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
@@ -1068,6 +1069,13 @@ RecordTransactionCommit(void)
XLogFlush(XactLastRecEnd);
/*
+ * Wake up all walsenders to send WAL up to the COMMIT record
+ * immediately if replication is enabled
+ */
+ if (max_wal_senders > 0)
+ WalSndWakeup();
+
+ /*
* Now we may update the CLOG, if we wrote a COMMIT record above
*/
if (markXidCommitted)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 53c2581..9f5d1af 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -66,9 +66,6 @@ bool am_walsender = false; /* Am I a walsender process ? */
int max_wal_senders = 0; /* the maximum number of concurrent walsenders */
int WalSndDelay = 200; /* max sleep time between some actions */
-#define NAPTIME_PER_CYCLE 100000L /* max sleep time between cycles
- * (100ms) */
-
/*
* These variables are used similarly to openLogFile/Id/Seg/Off,
* but for walsender to read the XLOG.
@@ -93,6 +90,7 @@ static volatile sig_atomic_t ready_to_stop = false;
static void WalSndSigHupHandler(SIGNAL_ARGS);
static void WalSndShutdownHandler(SIGNAL_ARGS);
static void WalSndQuickDieHandler(SIGNAL_ARGS);
+static void WalSndXLogSendHandler(SIGNAL_ARGS);
static void WalSndLastCycleHandler(SIGNAL_ARGS);
/* Prototypes for private functions */
@@ -144,6 +142,16 @@ WalSenderMain(void)
/* Handle handshake messages before streaming */
WalSndHandshake();
+ /* Initialize shared memory status */
+ {
+ /* use volatile pointer to prevent code rearrangement */
+ volatile WalSnd *walsnd = MyWalSnd;
+
+ SpinLockAcquire(&walsnd->mutex);
+ walsnd->sentPtr = sentPtr;
+ SpinLockRelease(&walsnd->mutex);
+ }
+
/* Main loop of walsender */
return WalSndLoop();
}
@@ -380,8 +388,6 @@ WalSndLoop(void)
/* Loop forever, unless we get an error */
for (;;)
{
- long remain; /* remaining time (us) */
-
/*
* Emergency bailout if postmaster has died. This is to avoid the
* necessity for manual cleanup of all postmaster children.
@@ -421,32 +427,41 @@ WalSndLoop(void)
/*
* If we had sent all accumulated WAL in last round, nap for the
* configured time before retrying.
- *
- * On some platforms, signals won't interrupt the sleep. To ensure we
- * respond reasonably promptly when someone signals us, break down the
- * sleep into NAPTIME_PER_CYCLE increments, and check for interrupts
- * after each nap.
*/
if (caughtup)
{
- remain = WalSndDelay * 1000L;
- while (remain > 0)
- {
- /* Check for interrupts */
- if (got_SIGHUP || shutdown_requested || ready_to_stop)
- break;
+ /*
+ * Even if we wrote all the WAL that was available when we started
+ * sending, more might have arrived while we were sending this
+ * batch. We had the latch set while sending, so we have not
+ * received any signals from that time. Let's arm the latch
+ * again, and after that check that we're still up-to-date.
+ */
+ ResetLatch(&MyWalSnd->latch);
- /* Sleep and check that the connection is still alive */
- pg_usleep(remain > NAPTIME_PER_CYCLE ? NAPTIME_PER_CYCLE : remain);
- CheckClosedConnection();
+ if (!XLogSend(output_message, &caughtup))
+ break;
+ if (caughtup && !got_SIGHUP && !ready_to_stop && !shutdown_requested)
+ {
+ /*
+ * XXX: Should we invent an API to wait for data coming from the
+ * client connection too? It's not critical, but we could then
+ * eliminate the timeout altogether and go to sleep for good.
+ */
- remain -= NAPTIME_PER_CYCLE;
+ /* Sleep */
+ WaitLatch(&MyWalSnd->latch, WalSndDelay * 1000);
}
- }
- /* Attempt to send the log once every loop */
- if (!XLogSend(output_message, &caughtup))
- break;
+ /* Check if the connection was closed */
+ CheckClosedConnection();
+ }
+ else
+ {
+ /* Attempt to send the log once every loop */
+ if (!XLogSend(output_message, &caughtup))
+ break;
+ }
}
/*
@@ -493,10 +508,15 @@ InitWalSnd(void)
}
else
{
- /* found */
- MyWalSnd = (WalSnd *) walsnd;
+ /*
+ * Found a free slot. Take ownership of the latch and initialize
+ * the other fields.
+ */
+ InitLatch((Latch *) &walsnd->latch);
walsnd->pid = MyProcPid;
- MemSet(&MyWalSnd->sentPtr, 0, sizeof(XLogRecPtr));
+ MemSet(&walsnd->sentPtr, 0, sizeof(XLogRecPtr));
+ /* Set MyWalSnd only after it's fully initialized. */
+ MyWalSnd = (WalSnd *) walsnd;
SpinLockRelease(&walsnd->mutex);
break;
}
@@ -523,6 +543,7 @@ WalSndKill(int code, Datum arg)
* for this.
*/
MyWalSnd->pid = 0;
+ ReleaseLatch(&MyWalSnd->latch);
/* WalSnd struct isn't mine anymore */
MyWalSnd = NULL;
@@ -787,6 +808,8 @@ static void
WalSndSigHupHandler(SIGNAL_ARGS)
{
got_SIGHUP = true;
+ if (MyWalSnd)
+ SetLatch(&MyWalSnd->latch);
}
/* SIGTERM: set flag to shut down */
@@ -794,6 +817,8 @@ static void
WalSndShutdownHandler(SIGNAL_ARGS)
{
shutdown_requested = true;
+ if (MyWalSnd)
+ SetLatch(&MyWalSnd->latch);
}
/*
@@ -828,11 +853,20 @@ WalSndQuickDieHandler(SIGNAL_ARGS)
exit(2);
}
+/* SIGUSR1: set flag to send WAL records */
+static void
+WalSndXLogSendHandler(SIGNAL_ARGS)
+{
+ latch_sigusr1_handler();
+}
+
/* SIGUSR2: set flag to do a last cycle and shut down afterwards */
static void
WalSndLastCycleHandler(SIGNAL_ARGS)
{
ready_to_stop = true;
+ if (MyWalSnd)
+ SetLatch(&MyWalSnd->latch);
}
/* Set up signal handlers */
@@ -847,7 +881,7 @@ WalSndSignals(void)
pqsignal(SIGQUIT, WalSndQuickDieHandler); /* hard crash time */
pqsignal(SIGALRM, SIG_IGN);
pqsignal(SIGPIPE, SIG_IGN);
- pqsignal(SIGUSR1, SIG_IGN); /* not used */
+ pqsignal(SIGUSR1, WalSndXLogSendHandler); /* request WAL sending */
pqsignal(SIGUSR2, WalSndLastCycleHandler); /* request a last cycle and
* shutdown */
@@ -895,6 +929,16 @@ WalSndShmemInit(void)
}
}
+/* Wake up all walsenders */
+void
+WalSndWakeup(void)
+{
+ int i;
+
+ for (i = 0; i < max_wal_senders; i++)
+ SetLatch(&WalSndCtl->walsnds[i].latch);
+}
+
/*
* This isn't currently used for anything. Monitoring tools might be
* interested in the future, and we'll need something like this in the
diff --git a/src/backend/storage/ipc/Makefile b/src/backend/storage/ipc/Makefile
index a5da0ec..5cbc515 100644
--- a/src/backend/storage/ipc/Makefile
+++ b/src/backend/storage/ipc/Makefile
@@ -15,7 +15,7 @@ override CFLAGS+= -fno-inline
endif
endif
-OBJS = ipc.o ipci.o pmsignal.o procarray.o procsignal.o shmem.o shmqueue.o \
- sinval.o sinvaladt.o standby.o
+OBJS = ipc.o ipci.o latch.o pmsignal.o procarray.o procsignal.o shmem.o \
+ shmqueue.o sinval.o sinvaladt.o standby.o
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 492ac9a..0083513 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -30,6 +30,7 @@
#include "replication/walsender.h"
#include "storage/bufmgr.h"
#include "storage/ipc.h"
+#include "storage/latch.h"
#include "storage/pg_shmem.h"
#include "storage/pmsignal.h"
#include "storage/procarray.h"
@@ -117,6 +118,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
size = add_size(size, SInvalShmemSize());
size = add_size(size, PMSignalShmemSize());
size = add_size(size, ProcSignalShmemSize());
+ size = add_size(size, LatchShmemSize());
size = add_size(size, BgWriterShmemSize());
size = add_size(size, AutoVacuumShmemSize());
size = add_size(size, WalSndShmemSize());
@@ -217,6 +219,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
*/
PMSignalShmemInit();
ProcSignalShmemInit();
+ LatchShmemInit();
BgWriterShmemInit();
AutoVacuumShmemInit();
WalSndShmemInit();
diff --git a/src/backend/storage/ipc/latch.c b/src/backend/storage/ipc/latch.c
new file mode 100644
index 0000000..1becf7a
--- /dev/null
+++ b/src/backend/storage/ipc/latch.c
@@ -0,0 +1,306 @@
+/*-------------------------------------------------------------------------
+ *
+ * latch.c
+ * Routines for interprocess latches
+ *
+ * A latch allows you to wait until another process, or a signal handler
+ * within the same process, wakes you up. There is three basic operations
+ * on a latch:
+ *
+ * SetLatch - Sets the latch
+ * ResetLatch - Clears the latch, allowing it to be set again
+ * WaitLatch - waits for the latch to become set
+ *
+ * These can be used to wait for an event, without the race conditions
+ * involved with e.g plain Unix signals and select(). pselect() was
+ * invented to solve the same problem, but it is not portable enough.
+ *
+ * The implementation is such that setting a latch that's already set
+ * is quick.
+ *
+ * The pattern to wait on an event is:
+ *
+ * for (;;)
+ * {
+ * WaitLatch();
+ * ResetLatch();
+ *
+ * if (work to do)
+ * Do Stuff();
+ * }
+ *
+ * It's important to reset the latch *before* checking if there's work to
+ * do. Otherwise, if someone sets the latch between the check and the
+ * ResetLatch call, you will miss it and Wait will block.
+ *
+ * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * $PostgreSQL$
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include <fcntl.h>
+#include <signal.h>
+#include <unistd.h>
+
+#include "miscadmin.h"
+#include "replication/walsender.h"
+#include "storage/latch.h"
+#include "storage/shmem.h"
+
+/* Read and write end of the self-pipe */
+static int selfpipe_readfd;
+static int selfpipe_writefd;
+
+/* Are we currently in WaitLatch()? The signal handler would like to know. */
+static volatile sig_atomic_t waiting = false;
+
+/* private function prototypes */
+static void drainSelfPipe(void);
+static void sendSelfPipeByte(void);
+
+/*
+ * Initialize a backend-local latch.
+ */
+void
+InitLatch(Latch *latch)
+{
+ Assert(latch->owner_pid == 0);
+ latch->owner_pid = MyProcPid;
+ latch->is_set = false;
+}
+
+/*
+ * Initialize an inter-process latch. Like InitLatch(), but the latch can
+ * be triggered from another process. A process that needs to wait on
+ * an inter-proess latch also needs to ensure that latch_sigusr1_handler()
+ * is called from the SIGUSR1 signal handler.
+ */
+void
+InitSharedLatch(Latch *latch)
+{
+ /*
+ * This is the same as InitLatch() in this implementation. The
+ * Windows implementation will likely differ.
+ */
+ InitLatch(latch);
+}
+
+/*
+ * Release a latch previously allocated with InitLatch() or InitShareLatch().
+ */
+void
+ReleaseLatch(Latch *latch)
+{
+ Assert(latch->owner_pid == MyProcPid);
+ latch->owner_pid = 0;
+}
+
+/*
+ * Wait for given latch to be set, or until 'timeout' milliseconds passes.
+ * If 'timeout' is 0, wait forever. If the latch is already set, returns
+ * immediately.
+ *
+ * The latch must have been previously initialized by the current process.
+ */
+void
+WaitLatch(Latch *latch, long timeout)
+{
+ struct timeval tv;
+ fd_set input_mask;
+ int rc;
+
+ if (latch->owner_pid != MyProcPid)
+ elog(ERROR, "cannot wait on a latch now owned by current process");
+
+ waiting = true;
+ for (;;)
+ {
+ /*
+ * Clear the pipe, and check if the latch is set already. If someone
+ * sets the latch between this and the select() below, the setter
+ * will write a byte to the pipe (or signal us and the signal handler
+ * will do that), and the select() will return immediately.
+ */
+ drainSelfPipe();
+ if (latch->is_set)
+ break;
+
+ /* Sleep */
+ if (timeout > 0)
+ {
+ tv.tv_sec = timeout / 1000000L;
+ tv.tv_usec = timeout % 1000000L;
+ }
+
+ FD_ZERO(&input_mask);
+ FD_SET(selfpipe_readfd, &input_mask);
+
+ rc = select(selfpipe_readfd + 1, &input_mask, NULL, NULL,
+ (timeout > 0) ? &tv : NULL);
+ if (rc < 0)
+ {
+ if (errno != EINTR)
+ ereport(ERROR,
+ (errcode_for_socket_access(),
+ errmsg("select() failed: %m")));
+ }
+ if (rc == 0)
+ break; /* timeout exceeded */
+ }
+ waiting = false;
+}
+
+/*
+ * Sets a latch and wakes up anyone waiting on it. Returns quickly if the
+ * latch is set already.
+ */
+void
+SetLatch(Latch *latch)
+{
+ pid_t owner_pid;
+
+ /* Quick exit if already set */
+ if (latch->is_set)
+ return;
+
+ latch->is_set = true;
+
+ /*
+ * See if anyone's waiting for the latch. It can be the current process
+ * if we're in a signal handler. We use the self-pipe to wake up the
+ * select() in that case. If it's another process, send a signal.
+ */
+ owner_pid = latch->owner_pid;
+ if (owner_pid == 0)
+ return;
+ else if (owner_pid == MyProcPid)
+ sendSelfPipeByte();
+ else
+ kill(owner_pid, SIGUSR1);
+}
+
+/*
+ * Clear the latch. Calling WaitLatch after this will sleep, unless
+ * the latch is set again before the WaitLatch call.
+ */
+void
+ResetLatch(Latch *latch)
+{
+ /* Only the owner should reset the latch */
+ Assert(latch->owner_pid == MyProcPid);
+
+ if (!latch->is_set)
+ return;
+
+ latch->is_set = false;
+}
+
+/*
+ * Unused in this implementation, but Windows will need this.
+ */
+static int
+NumSharedLatches(void)
+{
+ int numLatches = 0;
+
+ /* Each walsender needs one latch */
+ numLatches += max_wal_senders;
+
+ return numLatches;
+}
+
+/*
+ * LatchShmemSize
+ * Compute space needed for latch's shared memory
+ */
+Size
+LatchShmemSize(void)
+{
+ return 0;
+}
+
+/*
+ * LatchShmemInit
+ * Allocate and initialize shared memory needed for latches
+ */
+void
+LatchShmemInit(void)
+{
+ int pipefd[2];
+
+ /*
+ * Set up a pipe that allows a signal handler to wake up the select()
+ * in WaitLatch(). Make the write-end non-blocking, so that SetLatch()
+ * won't block if the event has already been set many times filling
+ * the kernel buffer. Make the read-end non-blocking too, so that we
+ * can easily clear the pipe by reading until EAGAIN or EWOULDBLOCK.
+ */
+ if (pipe(pipefd) < 0)
+ elog(FATAL, "pipe() failed: %m");
+ if (fcntl(pipefd[0], F_SETFL, O_NONBLOCK) < 0)
+ elog(FATAL, "fcntl() failed on read-end of self-pipe: %m");
+ if (fcntl(pipefd[1], F_SETFL, O_NONBLOCK) < 0)
+ elog(FATAL, "fcntl() failed on write-end of self-pipe: %m");
+
+ selfpipe_readfd = pipefd[0];
+ selfpipe_writefd = pipefd[1];
+}
+
+void
+latch_sigusr1_handler(void)
+{
+ if (waiting)
+ sendSelfPipeByte();
+}
+
+/* Send one byte to the self-pipe, to wake up WaitLatch() */
+static void
+sendSelfPipeByte(void)
+{
+ int rc;
+ char dummy = 0;
+
+retry:
+ rc = write(selfpipe_writefd, &dummy, 1);
+ if (rc < 0)
+ {
+ if (errno != EAGAIN && errno != EWOULDBLOCK)
+ {
+ /*
+ * XXX: Is it safe to elog(ERROR) in a signal handler?
+ */
+ elog(ERROR, "write() on self-pipe failed: %m");
+ }
+ if (errno == EINTR)
+ goto retry;
+ }
+}
+
+/* Read all available data from the self-pipe */
+static void
+drainSelfPipe(void)
+{
+ int rc;
+ char buf;
+
+ for (;;)
+ {
+ rc = read(selfpipe_readfd, &buf, 1);
+ if (rc < 0)
+ {
+ if (errno == EINTR)
+ continue;
+ if (errno == EAGAIN || errno == EWOULDBLOCK)
+ break; /* the pipe is empty */
+
+ elog(ERROR, "read() on self-pipe failed: %m");
+ }
+ else if (rc == 0)
+ elog(ERROR, "unexpected EOF on self-pipe");
+ }
+}
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 2340236..2ad2ad9 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -278,5 +278,7 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN))
RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
+ latch_sigusr1_handler();
+
errno = save_errno;
}
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 874959e..3a93820 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -12,7 +12,12 @@
#ifndef _WALSENDER_H
#define _WALSENDER_H
+#include "postgres.h"
+
+#include <signal.h>
+
#include "access/xlog.h"
+#include "storage/latch.h"
#include "storage/spin.h"
/*
@@ -23,6 +28,16 @@ typedef struct WalSnd
pid_t pid; /* this walsender's process id, or 0 */
XLogRecPtr sentPtr; /* WAL has been sent up to this point */
+ /* XXX
+ * When a walsender process is signaled, to wake it up to send any
+ * pending WAL, the sender of the signal should send busy to true.
+ * A walsender marked as busy should not be signaled again, to avoid
+ * redundant signaling which would slow down the walsender and the
+ * system as a whole. Walsender will clear the flag when it is
+ * finished sending all pending WAL again.
+ */
+ Latch latch;
+
slock_t mutex; /* locks shared variables shown above */
} WalSnd;
@@ -45,5 +60,6 @@ extern int WalSenderMain(void);
extern void WalSndSignals(void);
extern Size WalSndShmemSize(void);
extern void WalSndShmemInit(void);
+extern void WalSndWakeup(void);
#endif /* _WALSENDER_H */
diff --git a/src/include/storage/latch.h b/src/include/storage/latch.h
new file mode 100644
index 0000000..28989e5
--- /dev/null
+++ b/src/include/storage/latch.h
@@ -0,0 +1,38 @@
+/*-------------------------------------------------------------------------
+ *
+ * latch.h
+ * Routines for interprocess latches
+ *
+ *
+ * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * $PostgreSQL$
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef LATCH_H
+#define LATCH_H
+
+typedef struct
+{
+ volatile sig_atomic_t is_set;
+ volatile sig_atomic_t owner_pid;
+} Latch;
+
+/*
+ * prototypes for functions in latch.c
+ */
+extern void InitLatch(Latch *latch);
+extern void InitSharedLatch(Latch *latch);
+extern void ReleaseLatch(Latch *latch);
+extern void WaitLatch(Latch *latch, long timeout);
+extern void SetLatch(Latch *latch);
+extern void ResetLatch(Latch *latch);
+
+extern Size LatchShmemSize(void);
+extern void LatchShmemInit(void);
+
+extern void latch_sigusr1_handler(void);
+
+#endif /* LATCH_H */