diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 24c3dd32f8..eed778b779 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -2534,7 +2534,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,28 +2543,7 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) { 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(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++; + written = XLogWriteFile(openLogFile, from, nleft, startoffset); if (written <= 0) { @@ -2705,6 +2683,46 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible) } } +/* + * Issue pg_pwrite to write an XLOG 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. + */ +int +XLogWriteFile(int fd, const void *buf, size_t nbyte, off_t offset) +{ + int written; + instr_time start; + + /* 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(); + + /* + * 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/postmaster/checkpointer.c b/src/backend/postmaster/checkpointer.c index 3894f4a270..20454d4040 100644 --- a/src/backend/postmaster/checkpointer.c +++ b/src/backend/postmaster/checkpointer.c @@ -168,6 +168,7 @@ static bool IsCheckpointOnSchedule(double progress); static bool ImmediateCheckpointRequested(void); static bool CompactCheckpointerRequestQueue(void); static void UpdateSharedMemoryConfig(void); +static void pgstat_send_checkpointer(void); /* Signal handlers */ static void ReqCheckpointHandler(SIGNAL_ARGS); @@ -495,17 +496,8 @@ CheckpointerMain(void) /* Check for archive_timeout and switch xlog files if necessary. */ CheckArchiveTimeout(); - /* - * Send off activity statistics to the stats collector. (The reason - * why we re-use bgwriter-related code for this is that the bgwriter - * and checkpointer used to be just one process. It's probably not - * worth the trouble to split the stats support into two independent - * stats message types.) - */ - pgstat_send_bgwriter(); - - /* Send WAL statistics to the stats collector. */ - pgstat_report_wal(); + /* Send the statistics for the checkpointer to the stats collector */ + pgstat_send_checkpointer(); /* * If any checkpoint flags have been set, redo the loop to handle the @@ -572,8 +564,18 @@ HandleCheckpointerInterrupts(void) * back to the sigsetjmp block above */ ExitOnAnyError = true; - /* Close down the database */ + + /* + * Close down the database. + * + * Since ShutdownXLOG() creates restartpoint or checkpoint and updates + * the statistics, increment the checkpoint request and send the + * statistics to the stats collector. + */ + BgWriterStats.m_requested_checkpoints++; ShutdownXLOG(0, 0); + pgstat_send_checkpointer(); + /* Normal exit from the checkpointer is here */ proc_exit(0); /* done */ } @@ -1335,3 +1337,22 @@ FirstCallSinceLastCheckpoint(void) return FirstCall; } + +/* + * Send the statistics for the checkpointer to the stats collector + */ +static void +pgstat_send_checkpointer(void) +{ + /* + * Send off activity statistics to the stats collector. (The reason why + * we re-use bgwriter-related code for this is that the bgwriter and + * checkpointer used to be just one process. It's probably not worth the + * trouble to split the stats support into two independent stats message + * types.) + */ + pgstat_send_bgwriter(); + + /* Send WAL statistics to the stats collector. */ + pgstat_report_wal(); +} diff --git a/src/backend/postmaster/interrupt.c b/src/backend/postmaster/interrupt.c index dd9136a942..a50e00f06b 100644 --- a/src/backend/postmaster/interrupt.c +++ b/src/backend/postmaster/interrupt.c @@ -108,5 +108,7 @@ SignalHandlerForShutdownRequest(SIGNAL_ARGS) ShutdownRequestPending = true; SetLatch(MyLatch); + elog(DEBUG3, "received shutdown request signal"); + errno = save_errno; } diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index 68eefb9722..28fc3a80f4 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -329,6 +329,7 @@ static void pgstat_beshutdown_hook(int code, Datum arg); static PgStat_StatDBEntry *pgstat_get_db_entry(Oid databaseid, bool create); static PgStat_StatTabEntry *pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create); +static int pgstat_process_message(void); static void pgstat_write_statsfiles(bool permanent, bool allDbs); static void pgstat_write_db_statsfile(PgStat_StatDBEntry *dbentry, bool permanent); static HTAB *pgstat_read_statsfiles(Oid onlydb, bool permanent, bool deep); @@ -4810,8 +4811,6 @@ pgstat_send_slru(void) NON_EXEC_STATIC void PgstatCollectorMain(int argc, char *argv[]) { - int len; - PgStat_Msg msg; int wr; WaitEvent event; WaitEventSet *wes; @@ -4896,158 +4895,10 @@ PgstatCollectorMain(int argc, char *argv[]) pgstat_write_statsfiles(false, false); /* - * Try to receive and process a message. This will not block, - * since the socket is set to non-blocking mode. - * - * XXX On Windows, we have to force pgwin32_recv to cooperate, - * despite the previous use of pg_set_noblock() on the socket. - * This is extremely broken and should be fixed someday. + * Try to receive and process a message. */ -#ifdef WIN32 - pgwin32_noblock = 1; -#endif - - len = recv(pgStatSock, (char *) &msg, - sizeof(PgStat_Msg), 0); - -#ifdef WIN32 - pgwin32_noblock = 0; -#endif - - if (len < 0) - { - if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) - break; /* out of inner loop */ - ereport(ERROR, - (errcode_for_socket_access(), - errmsg("could not read statistics message: %m"))); - } - - /* - * We ignore messages that are smaller than our common header - */ - if (len < sizeof(PgStat_MsgHdr)) - continue; - - /* - * The received length must match the length in the header - */ - if (msg.msg_hdr.m_size != len) - continue; - - /* - * O.K. - we accept this message. Process it. - */ - switch (msg.msg_hdr.m_type) - { - case PGSTAT_MTYPE_DUMMY: - break; - - case PGSTAT_MTYPE_INQUIRY: - pgstat_recv_inquiry(&msg.msg_inquiry, len); - break; - - case PGSTAT_MTYPE_TABSTAT: - pgstat_recv_tabstat(&msg.msg_tabstat, len); - break; - - case PGSTAT_MTYPE_TABPURGE: - pgstat_recv_tabpurge(&msg.msg_tabpurge, len); - break; - - case PGSTAT_MTYPE_DROPDB: - pgstat_recv_dropdb(&msg.msg_dropdb, len); - break; - - case PGSTAT_MTYPE_RESETCOUNTER: - pgstat_recv_resetcounter(&msg.msg_resetcounter, len); - break; - - case PGSTAT_MTYPE_RESETSHAREDCOUNTER: - pgstat_recv_resetsharedcounter(&msg.msg_resetsharedcounter, - len); - break; - - case PGSTAT_MTYPE_RESETSINGLECOUNTER: - pgstat_recv_resetsinglecounter(&msg.msg_resetsinglecounter, - len); - break; - - case PGSTAT_MTYPE_RESETSLRUCOUNTER: - pgstat_recv_resetslrucounter(&msg.msg_resetslrucounter, - len); - break; - - case PGSTAT_MTYPE_RESETREPLSLOTCOUNTER: - pgstat_recv_resetreplslotcounter(&msg.msg_resetreplslotcounter, - len); - break; - - case PGSTAT_MTYPE_AUTOVAC_START: - pgstat_recv_autovac(&msg.msg_autovacuum_start, len); - break; - - case PGSTAT_MTYPE_VACUUM: - pgstat_recv_vacuum(&msg.msg_vacuum, len); - break; - - case PGSTAT_MTYPE_ANALYZE: - pgstat_recv_analyze(&msg.msg_analyze, len); - break; - - case PGSTAT_MTYPE_ARCHIVER: - pgstat_recv_archiver(&msg.msg_archiver, len); - break; - - case PGSTAT_MTYPE_BGWRITER: - pgstat_recv_bgwriter(&msg.msg_bgwriter, len); - break; - - case PGSTAT_MTYPE_WAL: - pgstat_recv_wal(&msg.msg_wal, len); - break; - - case PGSTAT_MTYPE_SLRU: - pgstat_recv_slru(&msg.msg_slru, len); - break; - - case PGSTAT_MTYPE_FUNCSTAT: - pgstat_recv_funcstat(&msg.msg_funcstat, len); - break; - - case PGSTAT_MTYPE_FUNCPURGE: - pgstat_recv_funcpurge(&msg.msg_funcpurge, len); - break; - - case PGSTAT_MTYPE_RECOVERYCONFLICT: - pgstat_recv_recoveryconflict(&msg.msg_recoveryconflict, - len); - break; - - case PGSTAT_MTYPE_DEADLOCK: - pgstat_recv_deadlock(&msg.msg_deadlock, len); - break; - - case PGSTAT_MTYPE_TEMPFILE: - pgstat_recv_tempfile(&msg.msg_tempfile, len); - break; - - case PGSTAT_MTYPE_CHECKSUMFAILURE: - pgstat_recv_checksum_failure(&msg.msg_checksumfailure, - len); - break; - - case PGSTAT_MTYPE_REPLSLOT: - pgstat_recv_replslot(&msg.msg_replslot, len); - break; - - case PGSTAT_MTYPE_CONNECTION: - pgstat_recv_connstat(&msg.msg_conn, len); - break; - - default: - break; - } + if (pgstat_process_message() < 0) + break; /* If an error occurred, go out of inner loop */ } /* end of inner message-processing loop */ /* Sleep until there's something to do */ @@ -5077,6 +4928,21 @@ PgstatCollectorMain(int argc, char *argv[]) break; } /* end of outer loop */ + /* + * Try to receive and process remaining messages before the process exits. + * + * The reason is that there is no guarantee all messages were processed in + * the above loop even though the stats collector is sent SIGQUIT signal + * by the postmaster after other backend and background processes, which + * sent their stats to the stats collector, exit if shutdown mode is smart + * or fast. + * + * For example, there might be a case that messages are lost when there + * are unprocessed messages, the postmaster send SIGQUIT signal to the + * stats collector. + */ + while (pgstat_process_message() > 0); + /* * Save the final stats to reuse at next startup. */ @@ -5225,6 +5091,171 @@ pgstat_get_tab_entry(PgStat_StatDBEntry *dbentry, Oid tableoid, bool create) return result; } +/* + * Try to receive and process a message. This will not block, + * since the socket is set to non-blocking mode. + * + * XXX On Windows, we have to force pgwin32_recv to cooperate, + * despite the previous use of pg_set_noblock() on the socket. + * This is extremely broken and should be fixed someday. + * + * Return the number of processed message. -1 if an error occurred. + */ +static int +pgstat_process_message() +{ + int len; + PgStat_Msg msg; + +#ifdef WIN32 + pgwin32_noblock = 1; +#endif + + len = recv(pgStatSock, (char *) &msg, sizeof(PgStat_Msg), 0); + +#ifdef WIN32 + pgwin32_noblock = 0; +#endif + + if (len < 0) + { + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) + return -1; + ereport(ERROR, + (errcode_for_socket_access(), + errmsg("could not read statistics message: %m"))); + } + + /* + * We ignore messages that are smaller than our common header + */ + if (len < sizeof(PgStat_MsgHdr)) + return 0; + + /* + * The received length must match the length in the header + */ + if (msg.msg_hdr.m_size != len) + return 0; + + /* + * O.K. - we accept this message. Process it. + */ + switch (msg.msg_hdr.m_type) + { + case PGSTAT_MTYPE_DUMMY: + break; + + case PGSTAT_MTYPE_INQUIRY: + pgstat_recv_inquiry(&msg.msg_inquiry, len); + break; + + case PGSTAT_MTYPE_TABSTAT: + pgstat_recv_tabstat(&msg.msg_tabstat, len); + break; + + case PGSTAT_MTYPE_TABPURGE: + pgstat_recv_tabpurge(&msg.msg_tabpurge, len); + break; + + case PGSTAT_MTYPE_DROPDB: + pgstat_recv_dropdb(&msg.msg_dropdb, len); + break; + + case PGSTAT_MTYPE_RESETCOUNTER: + pgstat_recv_resetcounter(&msg.msg_resetcounter, len); + break; + + case PGSTAT_MTYPE_RESETSHAREDCOUNTER: + pgstat_recv_resetsharedcounter(&msg.msg_resetsharedcounter, + len); + break; + + case PGSTAT_MTYPE_RESETSINGLECOUNTER: + pgstat_recv_resetsinglecounter(&msg.msg_resetsinglecounter, + len); + break; + + case PGSTAT_MTYPE_RESETSLRUCOUNTER: + pgstat_recv_resetslrucounter(&msg.msg_resetslrucounter, + len); + break; + + case PGSTAT_MTYPE_RESETREPLSLOTCOUNTER: + pgstat_recv_resetreplslotcounter(&msg.msg_resetreplslotcounter, + len); + break; + + case PGSTAT_MTYPE_AUTOVAC_START: + pgstat_recv_autovac(&msg.msg_autovacuum_start, len); + break; + + case PGSTAT_MTYPE_VACUUM: + pgstat_recv_vacuum(&msg.msg_vacuum, len); + break; + + case PGSTAT_MTYPE_ANALYZE: + pgstat_recv_analyze(&msg.msg_analyze, len); + break; + + case PGSTAT_MTYPE_ARCHIVER: + pgstat_recv_archiver(&msg.msg_archiver, len); + break; + + case PGSTAT_MTYPE_BGWRITER: + elog(DEBUG3, "process BGWRITER stats message"); + pgstat_recv_bgwriter(&msg.msg_bgwriter, len); + break; + + case PGSTAT_MTYPE_WAL: + elog(DEBUG3, "process WAL stats message"); + pgstat_recv_wal(&msg.msg_wal, len); + break; + + case PGSTAT_MTYPE_SLRU: + pgstat_recv_slru(&msg.msg_slru, len); + break; + + case PGSTAT_MTYPE_FUNCSTAT: + pgstat_recv_funcstat(&msg.msg_funcstat, len); + break; + + case PGSTAT_MTYPE_FUNCPURGE: + pgstat_recv_funcpurge(&msg.msg_funcpurge, len); + break; + + case PGSTAT_MTYPE_RECOVERYCONFLICT: + pgstat_recv_recoveryconflict(&msg.msg_recoveryconflict, + len); + break; + + case PGSTAT_MTYPE_DEADLOCK: + pgstat_recv_deadlock(&msg.msg_deadlock, len); + break; + + case PGSTAT_MTYPE_TEMPFILE: + pgstat_recv_tempfile(&msg.msg_tempfile, len); + break; + + case PGSTAT_MTYPE_CHECKSUMFAILURE: + pgstat_recv_checksum_failure(&msg.msg_checksumfailure, + len); + break; + + case PGSTAT_MTYPE_REPLSLOT: + pgstat_recv_replslot(&msg.msg_replslot, len); + break; + + case PGSTAT_MTYPE_CONNECTION: + pgstat_recv_connstat(&msg.msg_conn, len); + break; + + default: + break; + } + + return 1; +} /* ---------- * pgstat_write_statsfiles() - diff --git a/src/backend/postmaster/walwriter.c b/src/backend/postmaster/walwriter.c index 132df29aba..45c8531ac8 100644 --- a/src/backend/postmaster/walwriter.c +++ b/src/backend/postmaster/walwriter.c @@ -78,6 +78,9 @@ int WalWriterFlushAfter = 128; #define LOOPS_UNTIL_HIBERNATE 50 #define HIBERNATE_FACTOR 25 +/* Prototypes for private functions */ +static void HandleWalWriterInterrupts(void); + /* * Main entry point for walwriter process * @@ -242,7 +245,7 @@ WalWriterMain(void) /* Clear any already-pending wakeups */ ResetLatch(MyLatch); - HandleMainLoopInterrupts(); + HandleWalWriterInterrupts(); /* * Do what we're here for; then, if XLogBackgroundFlush() found useful @@ -272,3 +275,34 @@ WalWriterMain(void) WAIT_EVENT_WAL_WRITER_MAIN); } } + +/* + * interrupt handler for main loops of WAL writer processes. + */ +static void +HandleWalWriterInterrupts(void) +{ + if (ProcSignalBarrierPending) + ProcessProcSignalBarrier(); + + if (ConfigReloadPending) + { + ConfigReloadPending = false; + ProcessConfigFile(PGC_SIGHUP); + } + + if (ShutdownRequestPending) + { + /* + * Force to send remaining WAL statistics to the stats collector at + * process exits. + * + * Since pgstat_send_wal is invoked with 'force' is false in main loop + * to avoid overloading to the stats collector, there may exist unsent + * stats counters for the WAL writer. + */ + pgstat_send_wal(true); + + proc_exit(0); + } +} diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index 7810ee916c..3abd8ac93b 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -770,6 +770,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 || @@ -907,6 +910,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; @@ -928,7 +937,8 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr) /* OK to write the logs */ errno = 0; - byteswritten = pg_pwrite(recvFile, buf, segbytes, (off_t) startoff); + byteswritten = XLogWriteFile(recvFile, buf, segbytes, (off_t) startoff); + if (byteswritten <= 0) { char xlogfname[MAXFNAMELEN]; diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 1e53d9d4ca..b345de8a28 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -290,6 +290,7 @@ 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); extern void CheckXLogRemoved(XLogSegNo segno, TimeLineID tli); extern XLogSegNo XLogGetLastRemovedSegno(void);