From: Kyotaro Horiguchi Date: Thu, 21 Dec 2017 21:20:20 +0900 Subject: [PATCH 1/6] Add WAL relief vent for replication slots Replication slot is useful to maintain replication connection in the configurations where replication is so delayed that connection is broken. On the other hand so many WAL files can fill up disk that the master downs by a long delay. This feature, which is activated by a GUC "max_slot_wal_keep_size", protects master servers from suffering disk full by limiting the number of WAL files reserved by replication slots. --- src/backend/access/transam/xlog.c | 124 ++++++++++++++---- src/backend/replication/slot.c | 57 ++++++++ src/backend/utils/misc/guc.c | 12 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/access/xlog.h | 1 + src/include/replication/slot.h | 1 + 6 files changed, 173 insertions(+), 23 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index e08320e829..9bea2a403f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -104,6 +104,7 @@ int wal_level = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; +int max_slot_wal_keep_size_mb = -1; #ifdef WAL_DEBUG bool XLOG_DEBUG = false; @@ -871,6 +872,7 @@ static void checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI, static void LocalSetXLogInsertAllowed(void); static void CreateEndOfRecoveryRecord(void); static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags); +static XLogSegNo GetOldestKeepSegment(XLogRecPtr currpos, XLogRecPtr minSlotPtr); static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo); static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void); @@ -9272,6 +9274,54 @@ CreateRestartPoint(int flags) return true; } +/* + * Returns minimum segment number that the next checkpoint must leave + * considering wal_keep_segments, replication slots and + * max_slot_wal_keep_size. + * + * currLSN is the current insert location. + * minSlotLSN is the minimum restart_lsn of all active slots. + */ +static XLogSegNo +GetOldestKeepSegment(XLogRecPtr currLSN, XLogRecPtr minSlotLSN) +{ + XLogSegNo currSeg; + XLogSegNo minSlotSeg; + uint64 keepSegs = 0; /* # of segments actually kept */ + + XLByteToSeg(currLSN, currSeg, wal_segment_size); + XLByteToSeg(minSlotLSN, minSlotSeg, wal_segment_size); + + /* + * Calculate how many segments are kept by slots first. The second + * term of the condition is just a sanity check. + */ + if (minSlotLSN != InvalidXLogRecPtr && minSlotSeg <= currSeg) + keepSegs = currSeg - minSlotSeg; + + /* Cap keepSegs by max_slot_wal_keep_size */ + if (max_slot_wal_keep_size_mb >= 0) + { + uint64 limitSegs; + + limitSegs = ConvertToXSegs(max_slot_wal_keep_size_mb, wal_segment_size); + + /* Reduce it if slots already reserves too many. */ + if (limitSegs < keepSegs) + keepSegs = limitSegs; + } + + /* but, keep at least wal_keep_segments segments if any */ + if (wal_keep_segments > 0 && keepSegs < wal_keep_segments) + keepSegs = wal_keep_segments; + + /* avoid underflow, don't go below 1 */ + if (currSeg <= keepSegs) + return 1; + + return currSeg - keepSegs; +} + /* * Retreat *logSegNo to the last segment that we need to retain because of * either wal_keep_segments or replication slots. @@ -9283,38 +9333,66 @@ CreateRestartPoint(int flags) static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo) { - XLogSegNo segno; - XLogRecPtr keep; + XLogRecPtr slotminptr = InvalidXLogRecPtr; + XLogSegNo minSegNo; + XLogSegNo slotSegNo; - XLByteToSeg(recptr, segno, wal_segment_size); - keep = XLogGetReplicationSlotMinimumLSN(); + if (max_replication_slots > 0) + slotminptr = XLogGetReplicationSlotMinimumLSN(); - /* compute limit for wal_keep_segments first */ - if (wal_keep_segments > 0) - { - /* avoid underflow, don't go below 1 */ - if (segno <= wal_keep_segments) - segno = 1; - else - segno = segno - wal_keep_segments; - } + /* + * We should keep certain number of WAL segments after this checkpoint. + */ + minSegNo = GetOldestKeepSegment(recptr, slotminptr); - /* then check whether slots limit removal further */ - if (max_replication_slots > 0 && keep != InvalidXLogRecPtr) + /* + * Warn the checkpoint is going to flush the segments required by + * replication slots. + */ + if (!XLogRecPtrIsInvalid(slotminptr)) { - XLogSegNo slotSegNo; + static XLogSegNo prev_lost_segs = 0; /* avoid duplicate messages */ - XLByteToSeg(keep, slotSegNo, wal_segment_size); + XLByteToSeg(slotminptr, slotSegNo, wal_segment_size); - if (slotSegNo <= 0) - segno = 1; - else if (slotSegNo < segno) - segno = slotSegNo; + if (slotSegNo < minSegNo) + { + XLogSegNo lost_segs = minSegNo - slotSegNo; + if (prev_lost_segs != lost_segs) + { + /* We have lost a new segment, warn it.*/ + XLogRecPtr minlsn; + char *slot_names; + int nslots; + + XLogSegNoOffsetToRecPtr(minSegNo, 0, wal_segment_size, minlsn); + slot_names = + ReplicationSlotsEnumerateBehinds(minlsn, ", ", &nslots); + + /* + * Some of the affected slots could have just been removed. + * We don't need show anything here if no affected slot + * remains. + */ + if (slot_names) + { + ereport(WARNING, + (errmsg ("some replication slots have lost required WAL segments"), + errdetail_plural( + "Slot %s lost %ld segment(s).", + "Slots %s lost at most %ld segment(s).", + nslots, slot_names, lost_segs))); + } + } + prev_lost_segs = lost_segs; + } + else + prev_lost_segs = 0; } /* don't delete WAL segments newer than the calculated segment */ - if (segno < *logSegNo) - *logSegNo = segno; + if (minSegNo < *logSegNo) + *logSegNo = minSegNo; } /* diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 55c306e465..7f9eab6edf 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1064,6 +1064,63 @@ ReplicationSlotReserveWal(void) } } +/* + * Returns names of replication slots that their restart_lsn are behind + * specified LSN, in palloc'ed character array stuffed with slot names + * delimited by the given separator. Returns NULL if no slot matches. If + * pnslots is given, the number of the returned slots is returned there. + */ +char * +ReplicationSlotsEnumerateBehinds(XLogRecPtr target, char *separator, int *pnslots) +{ + static StringInfoData retstr; + static bool retstr_initialized = false; + bool insert_separator = false; + int i; + int nslots = 0; + + Assert (separator); + if (max_replication_slots <= 0) + return NULL; + + if (!retstr_initialized) + { + initStringInfo(&retstr); + retstr_initialized = true; + } + else + resetStringInfo(&retstr); + + /* construct name list */ + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); + for (i = 0 ; i < max_replication_slots ; i++) + { + ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i]; + + if (s->in_use && s->data.restart_lsn < target) + { + if (insert_separator) + appendStringInfoString(&retstr, separator); + + /* + * Slot names consist only with lower-case letters. We don't + * bother quoting. + */ + appendStringInfoString(&retstr, NameStr(s->data.name)); + insert_separator = true; + nslots++; + } + } + LWLockRelease(ReplicationSlotControlLock); + + /* return the number of slots in the list if requested */ + if (pnslots) + *pnslots = nslots; + + /* return NULL instead of an empty string */ + return retstr.data[0] ? retstr.data : NULL; +} + /* * Flush all replication slots to disk. * diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 1208eb9a68..80245e9def 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -2655,6 +2655,18 @@ static struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"max_slot_wal_keep_size", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Sets the maximum size of extra WALs kept by replication slots."), + NULL, + GUC_UNIT_MB + }, + &max_slot_wal_keep_size_mb, + -1, -1, + MAX_KILOBYTES, /* XXX: This is in megabytes, like max/min_wal_size */ + NULL, NULL, NULL + }, + { {"wal_sender_timeout", PGC_USERSET, REPLICATION_SENDING, gettext_noop("Sets the maximum time to wait for WAL replication."), diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 5ee5e09ddf..5da2706a99 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -286,6 +286,7 @@ #max_wal_senders = 10 # max number of walsender processes # (change requires restart) #wal_keep_segments = 0 # in logfile segments; 0 disables +#max_slot_wal_keep_size = -1 # measured in bytes; -1 disables #wal_sender_timeout = 60s # in milliseconds; 0 disables #max_replication_slots = 10 # max number of replication slots diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index 237f4e0350..e5322abdf5 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -108,6 +108,7 @@ extern int wal_segment_size; extern int min_wal_size_mb; extern int max_wal_size_mb; extern int wal_keep_segments; +extern int max_slot_wal_keep_size_mb; extern int XLOGbuffers; extern int XLogArchiveTimeout; extern int wal_retrieve_retry_interval; diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 8fbddea78f..e0fee0663c 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -199,6 +199,7 @@ extern void ReplicationSlotsComputeRequiredLSN(void); extern XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void); extern bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive); extern void ReplicationSlotsDropDBSlots(Oid dboid); +extern char *ReplicationSlotsEnumerateBehinds(XLogRecPtr target, char *separator, int *pnslots); extern void StartupReplicationSlots(void); extern void CheckPointReplicationSlots(void); -- 2.20.1 --MP_/68vU=6Ib.yoYOTdA.M=EVne Content-Type: text/x-patch Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename=v14-0002-Add-monitoring-aid-for-max_slot_wal_keep_size.patch