Initial commit

vendor/github.com/cockroachdb/pebble/pacer.go

@@ -0,0 +1,190 @@
+// Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
+// of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package pebble
+
+import (
+	"sync"
+	"time"
+)
+
+// deletionPacerInfo contains any info from the db necessary to make deletion
+// pacing decisions (to limit background IO usage so that it does not contend
+// with foreground traffic).
+type deletionPacerInfo struct {
+	freeBytes     uint64
+	obsoleteBytes uint64
+	liveBytes     uint64
+}
+
+// deletionPacer rate limits deletions of obsolete files. This is necessary to
+// prevent overloading the disk with too many deletions too quickly after a
+// large compaction, or an iterator close. On some SSDs, disk performance can be
+// negatively impacted if too many blocks are deleted very quickly, so this
+// mechanism helps mitigate that.
+type deletionPacer struct {
+	// If there are less than freeSpaceThreshold bytes of free space on
+	// disk, increase the pace of deletions such that we delete enough bytes to
+	// get back to the threshold within the freeSpaceTimeframe.
+	freeSpaceThreshold uint64
+	freeSpaceTimeframe time.Duration
+
+	// If the ratio of obsolete bytes to live bytes is greater than
+	// obsoleteBytesMaxRatio, increase the pace of deletions such that we delete
+	// enough bytes to get back to the threshold within the obsoleteBytesTimeframe.
+	obsoleteBytesMaxRatio  float64
+	obsoleteBytesTimeframe time.Duration
+
+	mu struct {
+		sync.Mutex
+
+		// history keeps rack of recent deletion history; it used to increase the
+		// deletion rate to match the pace of deletions.
+		history history
+	}
+
+	targetByteDeletionRate int64
+
+	getInfo func() deletionPacerInfo
+}
+
+const deletePacerHistory = 5 * time.Minute
+
+// newDeletionPacer instantiates a new deletionPacer for use when deleting
+// obsolete files.
+//
+// targetByteDeletionRate is the rate (in bytes/sec) at which we want to
+// normally limit deletes (when we are not falling behind or running out of
+// space). A value of 0.0 disables pacing.
+func newDeletionPacer(
+	now time.Time, targetByteDeletionRate int64, getInfo func() deletionPacerInfo,
+) *deletionPacer {
+	d := &deletionPacer{
+		freeSpaceThreshold: 16 << 30, // 16 GB
+		freeSpaceTimeframe: 10 * time.Second,
+
+		obsoleteBytesMaxRatio:  0.20,
+		obsoleteBytesTimeframe: 5 * time.Minute,
+
+		targetByteDeletionRate: targetByteDeletionRate,
+		getInfo:                getInfo,
+	}
+	d.mu.history.Init(now, deletePacerHistory)
+	return d
+}
+
+// ReportDeletion is used to report a deletion to the pacer. The pacer uses it
+// to keep track of the recent rate of deletions and potentially increase the
+// deletion rate accordingly.
+//
+// ReportDeletion is thread-safe.
+func (p *deletionPacer) ReportDeletion(now time.Time, bytesToDelete uint64) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.mu.history.Add(now, int64(bytesToDelete))
+}
+
+// PacingDelay returns the recommended pacing wait time (in seconds) for
+// deleting the given number of bytes.
+//
+// PacingDelay is thread-safe.
+func (p *deletionPacer) PacingDelay(now time.Time, bytesToDelete uint64) (waitSeconds float64) {
+	if p.targetByteDeletionRate == 0 {
+		// Pacing disabled.
+		return 0.0
+	}
+
+	baseRate := float64(p.targetByteDeletionRate)
+	// If recent deletion rate is more than our target, use that so that we don't
+	// fall behind.
+	historicRate := func() float64 {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+		return float64(p.mu.history.Sum(now)) / deletePacerHistory.Seconds()
+	}()
+	if historicRate > baseRate {
+		baseRate = historicRate
+	}
+
+	// Apply heuristics to increase the deletion rate.
+	var extraRate float64
+	info := p.getInfo()
+	if info.freeBytes <= p.freeSpaceThreshold {
+		// Increase the rate so that we can free up enough bytes within the timeframe.
+		extraRate = float64(p.freeSpaceThreshold-info.freeBytes) / p.freeSpaceTimeframe.Seconds()
+	}
+	if info.liveBytes == 0 {
+		// We don't know the obsolete bytes ratio. Disable pacing altogether.
+		return 0.0
+	}
+	obsoleteBytesRatio := float64(info.obsoleteBytes) / float64(info.liveBytes)
+	if obsoleteBytesRatio >= p.obsoleteBytesMaxRatio {
+		// Increase the rate so that we can free up enough bytes within the timeframe.
+		r := (obsoleteBytesRatio - p.obsoleteBytesMaxRatio) * float64(info.liveBytes) / p.obsoleteBytesTimeframe.Seconds()
+		if extraRate < r {
+			extraRate = r
+		}
+	}
+
+	return float64(bytesToDelete) / (baseRate + extraRate)
+}
+
+// history is a helper used to keep track of the recent history of a set of
+// data points (in our case deleted bytes), at limited granularity.
+// Specifically, we split the desired timeframe into 100 "epochs" and all times
+// are effectively rounded down to the nearest epoch boundary.
+type history struct {
+	epochDuration time.Duration
+	startTime     time.Time
+	// currEpoch is the epoch of the most recent operation.
+	currEpoch int64
+	// val contains the recent epoch values.
+	// val[currEpoch % historyEpochs] is the current epoch.
+	// val[(currEpoch + 1) % historyEpochs] is the oldest epoch.
+	val [historyEpochs]int64
+	// sum is always equal to the sum of values in val.
+	sum int64
+}
+
+const historyEpochs = 100
+
+// Init the history helper to keep track of data over the given number of
+// seconds.
+func (h *history) Init(now time.Time, timeframe time.Duration) {
+	*h = history{
+		epochDuration: timeframe / time.Duration(historyEpochs),
+		startTime:     now,
+		currEpoch:     0,
+		sum:           0,
+	}
+}
+
+// Add adds a value for the current time.
+func (h *history) Add(now time.Time, val int64) {
+	h.advance(now)
+	h.val[h.currEpoch%historyEpochs] += val
+	h.sum += val
+}
+
+// Sum returns the sum of recent values. The result is approximate in that the
+// cut-off time is within 1% of the exact one.
+func (h *history) Sum(now time.Time) int64 {
+	h.advance(now)
+	return h.sum
+}
+
+func (h *history) epoch(t time.Time) int64 {
+	return int64(t.Sub(h.startTime) / h.epochDuration)
+}
+
+// advance advances the time to the given time.
+func (h *history) advance(now time.Time) {
+	epoch := h.epoch(now)
+	for h.currEpoch < epoch {
+		h.currEpoch++
+		// Forget the data for the oldest epoch.
+		h.sum -= h.val[h.currEpoch%historyEpochs]
+		h.val[h.currEpoch%historyEpochs] = 0
+	}
+}