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 +#include +#include + +#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 + #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 */