diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 9b2eb0d10b..c7bda9127f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -2536,7 +2536,6 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) Size nbytes; Size nleft; int written; - instr_time start; /* OK to write the page(s) */ from = XLogCtl->pages + startidx * (Size) XLOG_BLCKSZ; @@ -2544,49 +2543,9 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) nleft = nbytes; do { - errno = 0; + written = XLogWriteFile(openLogFile, from, nleft, (off_t) startoffset, + ThisTimeLineID, openLogSegNo, wal_segment_size); - /* Measure I/O timing to write WAL data */ - if (track_wal_io_timing) - INSTR_TIME_SET_CURRENT(start); - - pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE); - written = pg_pwrite(openLogFile, from, nleft, startoffset); - pgstat_report_wait_end(); - - /* - * Increment the I/O timing and the number of times WAL data - * were written out to disk. - */ - if (track_wal_io_timing) - { - instr_time duration; - - INSTR_TIME_SET_CURRENT(duration); - INSTR_TIME_SUBTRACT(duration, start); - WalStats.m_wal_write_time += INSTR_TIME_GET_MICROSEC(duration); - } - - WalStats.m_wal_write++; - - if (written <= 0) - { - char xlogfname[MAXFNAMELEN]; - int save_errno; - - if (errno == EINTR) - continue; - - save_errno = errno; - XLogFileName(xlogfname, ThisTimeLineID, openLogSegNo, - wal_segment_size); - errno = save_errno; - ereport(PANIC, - (errcode_for_file_access(), - errmsg("could not write to log file %s " - "at offset %u, length %zu: %m", - xlogfname, startoffset, nleft))); - } nleft -= written; from += written; startoffset += written; @@ -2707,6 +2666,82 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) } } +/* + * Issue pg_pwrite to write an WAL segment file. + * + * 'fd' is a file descriptor for the XLOG file to write + * 'buf' is a buffer starting address to write. + * 'nbyte' is a number of max bytes to write up. + * 'offset' is a offset of XLOG file to be set. + * 'timelineid' is a timeline ID of WAL segment file. + * 'segno' is a segment number of WAL segment file. + * 'segsize' is a segment size of WAL segment file. + * + * Return the number of bytes written. + */ +int +XLogWriteFile(int fd, const void *buf, size_t nbyte, off_t offset, + TimeLineID timelineid, XLogSegNo segno, int segsize) +{ + /* + * Loop until to write the buffer data or an error occurred. + */ + for (;;) + { + int written; + instr_time start; + + errno = 0; + + /* Measure I/O timing to write WAL data */ + if (track_wal_io_timing) + INSTR_TIME_SET_CURRENT(start); + + pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE); + written = pg_pwrite(fd, buf, nbyte, offset); + pgstat_report_wait_end(); + + if (written <= 0) + { + char xlogfname[MAXFNAMELEN]; + int save_errno; + + if (errno == EINTR) + continue; + + /* if write didn't set errno, assume no disk space */ + if (errno == 0) + errno = ENOSPC; + + save_errno = errno; + XLogFileName(xlogfname, timelineid, segno, segsize); + errno = save_errno; + ereport(PANIC, + (errcode_for_file_access(), + errmsg("could not write to log file %s " + "at offset %u, length %zu: %m", + xlogfname, (unsigned int) offset, (unsigned long) nbyte))); + } + + /* + * Increment the I/O timing and the number of times WAL data were + * written out to disk. + */ + if (track_wal_io_timing) + { + instr_time duration; + + INSTR_TIME_SET_CURRENT(duration); + INSTR_TIME_SUBTRACT(duration, start); + WalStats.m_wal_write_time += INSTR_TIME_GET_MICROSEC(duration); + } + + WalStats.m_wal_write++; + + return written; + } +} + /* * Record the LSN for an asynchronous transaction commit/abort * and nudge the WALWriter if there is work for it to do. diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index a7a94d2a83..e9de78ffaa 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -771,6 +771,9 @@ WalRcvDie(int code, Datum arg) /* Ensure that all WAL records received are flushed to disk */ XLogWalRcvFlush(true); + /* Send WAL statistics to the stats collector before terminating */ + pgstat_send_wal(true); + /* Mark ourselves inactive in shared memory */ SpinLockAcquire(&walrcv->mutex); Assert(walrcv->walRcvState == WALRCV_STREAMING || @@ -868,7 +871,7 @@ XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len) static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) { - int startoff; + uint32 startoff; int byteswritten; while (nbytes > 0) @@ -910,6 +913,12 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) XLogArchiveForceDone(xlogfname); else XLogArchiveNotify(xlogfname); + + /* + * Send WAL statistics to the stats collector when finishing + * the current WAL segment file to avoid overloading it. + */ + pgstat_send_wal(false); } recvFile = -1; @@ -929,27 +938,8 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) segbytes = nbytes; /* OK to write the logs */ - errno = 0; - - byteswritten = pg_pwrite(recvFile, buf, segbytes, (off_t) startoff); - if (byteswritten <= 0) - { - char xlogfname[MAXFNAMELEN]; - int save_errno; - - /* if write didn't set errno, assume no disk space */ - if (errno == 0) - errno = ENOSPC; - - save_errno = errno; - XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size); - errno = save_errno; - ereport(PANIC, - (errcode_for_file_access(), - errmsg("could not write to log segment %s " - "at offset %u, length %lu: %m", - xlogfname, startoff, (unsigned long) segbytes))); - } + byteswritten = XLogWriteFile(recvFile, buf, segbytes, (off_t) startoff, + recvFileTLI, recvSegNo, wal_segment_size); /* Update state for write */ recptr += byteswritten; diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 6d384d3ce6..fd478b3478 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -298,6 +298,10 @@ extern bool XLogBackgroundFlush(void); extern bool XLogNeedsFlush(XLogRecPtr RecPtr); extern int XLogFileInit(XLogSegNo segno, bool *use_existent, bool use_lock); extern int XLogFileOpen(XLogSegNo segno); +extern int XLogWriteFile(int fd, const void *buf, + size_t nbyte, off_t offset, + TimeLineID timelineid, XLogSegNo segno, + int segsize); extern void CheckXLogRemoved(XLogSegNo segno, TimeLineID tli); extern XLogSegNo XLogGetLastRemovedSegno(void);