*** a/src/backend/access/transam/twophase.c --- b/src/backend/access/transam/twophase.c *************** *** 55,60 **** --- 55,61 ---- #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,1030 **** EndPrepare(GlobalTransaction gxact) --- 1026,1038 ---- /* 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,2010 **** RecordTransactionCommitPrepared(TransactionId xid, --- 2013,2025 ---- /* 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,2083 **** RecordTransactionAbortPrepared(TransactionId xid, --- 2093,2105 ---- 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. */ *** a/src/backend/access/transam/xact.c --- b/src/backend/access/transam/xact.c *************** *** 36,41 **** --- 36,42 ---- #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,1073 **** RecordTransactionCommit(void) --- 1069,1081 ---- 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) *** a/src/backend/replication/walsender.c --- b/src/backend/replication/walsender.c *************** *** 87,98 **** static XLogRecPtr sentPtr = {0, 0}; --- 87,100 ---- /* Flags set by signal handlers for later service in main loop */ static volatile sig_atomic_t got_SIGHUP = false; static volatile sig_atomic_t shutdown_requested = false; + static volatile sig_atomic_t xlogsend_requested = false; static volatile sig_atomic_t ready_to_stop = false; /* Signal handlers */ 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 */ *************** *** 418,423 **** WalSndLoop(void) --- 420,428 ---- proc_exit(0); } + if (xlogsend_requested) + xlogsend_requested = false; + /* * If we had sent all accumulated WAL in last round, nap for the * configured time before retrying. *************** *** 433,439 **** WalSndLoop(void) while (remain > 0) { /* Check for interrupts */ ! if (got_SIGHUP || shutdown_requested || ready_to_stop) break; /* Sleep and check that the connection is still alive */ --- 438,445 ---- while (remain > 0) { /* Check for interrupts */ ! if (got_SIGHUP || shutdown_requested || xlogsend_requested || ! ready_to_stop) break; /* Sleep and check that the connection is still alive */ *************** *** 665,670 **** XLogSend(char *msgbuf, bool *caughtup) --- 671,678 ---- Size nbytes; WalDataMessageHeader msghdr; + MyWalSnd->sndrqst = false; + /* * Attempt to send all data that's already been written out and fsync'd to * disk. We cannot go further than what's been written out given the *************** *** 678,684 **** XLogSend(char *msgbuf, bool *caughtup) /* Quick exit if nothing to do */ if (XLByteLE(SendRqstPtr, sentPtr)) { ! *caughtup = true; return true; } --- 686,692 ---- /* Quick exit if nothing to do */ if (XLByteLE(SendRqstPtr, sentPtr)) { ! *caughtup = !MyWalSnd->sndrqst; return true; } *************** *** 718,730 **** XLogSend(char *msgbuf, bool *caughtup) if (XLByteLE(SendRqstPtr, endptr)) { endptr = SendRqstPtr; ! *caughtup = true; } else { /* round down to page boundary. */ endptr.xrecoff -= (endptr.xrecoff % XLOG_BLCKSZ); *caughtup = false; } nbytes = endptr.xrecoff - startptr.xrecoff; --- 726,739 ---- if (XLByteLE(SendRqstPtr, endptr)) { endptr = SendRqstPtr; ! *caughtup = !MyWalSnd->sndrqst; } else { /* round down to page boundary. */ endptr.xrecoff -= (endptr.xrecoff % XLOG_BLCKSZ); *caughtup = false; + MyWalSnd->sndrqst = true; } nbytes = endptr.xrecoff - startptr.xrecoff; *************** *** 828,833 **** WalSndQuickDieHandler(SIGNAL_ARGS) --- 837,849 ---- exit(2); } + /* SIGUSR1: set flag to send WAL records */ + static void + WalSndXLogSendHandler(SIGNAL_ARGS) + { + xlogsend_requested = true; + } + /* SIGUSR2: set flag to do a last cycle and shut down afterwards */ static void WalSndLastCycleHandler(SIGNAL_ARGS) *************** *** 847,853 **** WalSndSignals(void) pqsignal(SIGQUIT, WalSndQuickDieHandler); /* hard crash time */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); ! pqsignal(SIGUSR1, SIG_IGN); /* not used */ pqsignal(SIGUSR2, WalSndLastCycleHandler); /* request a last cycle and * shutdown */ --- 863,869 ---- pqsignal(SIGQUIT, WalSndQuickDieHandler); /* hard crash time */ pqsignal(SIGALRM, SIG_IGN); pqsignal(SIGPIPE, SIG_IGN); ! pqsignal(SIGUSR1, WalSndXLogSendHandler); /* request WAL sending */ pqsignal(SIGUSR2, WalSndLastCycleHandler); /* request a last cycle and * shutdown */ *************** *** 895,900 **** WalSndShmemInit(void) --- 911,937 ---- } } + /* Wake all walsenders up by signaling */ + void + WalSndWakeup(void) + { + int i; + + for (i = 0; i < max_wal_senders; i++) + { + /* use volatile pointer to prevent code rearrangement */ + volatile WalSnd *walsnd = &WalSndCtl->walsnds[i]; + pid_t pid = walsnd->pid; + + /* we don't need to signal to waking walsenders */ + if (pid != 0 && !walsnd->sndrqst) + { + walsnd->sndrqst = true; + kill(pid, SIGUSR1); + } + } + } + /* * This isn't currently used for anything. Monitoring tools might be * interested in the future, and we'll need something like this in the *** a/src/include/replication/walsender.h --- b/src/include/replication/walsender.h *************** *** 12,17 **** --- 12,21 ---- #ifndef _WALSENDER_H #define _WALSENDER_H + #include "postgres.h" + + #include + #include "access/xlog.h" #include "storage/spin.h" *************** *** 23,28 **** typedef struct WalSnd --- 27,42 ---- pid_t pid; /* this walsender's process id, or 0 */ XLogRecPtr sentPtr; /* WAL has been sent up to this point */ + /* + * The request for WAL sending has already been sent? If false, + * we set sndrqst to true and signal walsender to send WAL up to + * the current WAL write location immeidately. If true, since + * walsender is ready for sending all the currently-written WAL, + * we don't need to signal. This value is used to suppress + * redundant signaling. + */ + sig_atomic_t sndrqst; + slock_t mutex; /* locks shared variables shown above */ } WalSnd; *************** *** 45,49 **** extern int WalSenderMain(void); --- 59,64 ---- extern void WalSndSignals(void); extern Size WalSndShmemSize(void); extern void WalSndShmemInit(void); + extern void WalSndWakeup(void); #endif /* _WALSENDER_H */