From 8d388acc0e923a47897030b3cdbd38b9b81b2561 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 16 Jun 2026 09:16:24 -0700 Subject: [PATCH] mount: tolerance-window write pattern detection for concurrent writeback (#9984) * mount: tolerance-window write pattern detection for concurrent writeback WriterPattern decides whether each dirty chunk is buffered in RAM (NewMemChunk) or spilled to an on-disk swap file (NewSwapFileChunk), so a sequential stream misclassified as random is pushed through disk-backed swap. The old detector compared each write's start to the previous write's exact stop offset (atomic.Swap of lastWriteStopOffset, then lastOffset == offset). That is brittle under concurrent FUSE writeback: MonitorWriteAt runs before the per-handle write lock, so parallel Write upcalls interleave the swap, and writeback flushes dirty pages slightly out of offset order. A genuinely sequential stream then repeatedly reads as random and spills to swap. Replace the exact match with the same frontier + tolerance approach used on the read side: track a never-regressing max frontier (offset+size) via a CAS loop and treat a write whose start is within SeqTolerance (8 MiB) of that frontier as sequential. The existing +/-ModeChangeLimit hysteresis is kept, so a single outlier write can't flip the mode. Adds page_writer_pattern_test.go covering sequentiality, hysteresis, reorder tolerance, the inclusive tolerance boundary, frontier non-regression (the CAS invariant), and recovery back to sequential mode. * mount: read write frontier inside the CAS loop Capture the frontier from the CAS loop's own load rather than a separate up-front snapshot, so the sequentiality diff is judged against the freshest pre-image even if a concurrent writeback upcall advances the frontier while we loop. Also drops the now-redundant load. Behavior is unchanged for the single-threaded tests; addresses review feedback on #9984. --- weed/mount/page_writer_pattern.go | 40 ++++++++-- weed/mount/page_writer_pattern_test.go | 104 +++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 6 deletions(-) create mode 100644 weed/mount/page_writer_pattern_test.go diff --git a/weed/mount/page_writer_pattern.go b/weed/mount/page_writer_pattern.go index d0c3c417ca..2f79caf182 100644 --- a/weed/mount/page_writer_pattern.go +++ b/weed/mount/page_writer_pattern.go @@ -4,27 +4,55 @@ import "sync/atomic" type WriterPattern struct { isSequentialCounter int64 - lastWriteStopOffset int64 + writeFrontier int64 // highest (offset+size) observed across writes chunkSize int64 } const ModeChangeLimit = 3 -// For streaming write: only cache the first chunk -// For random write: fall back to temp file approach +// SeqTolerance: a write whose start is within this many bytes of the current +// write frontier still counts as sequential. Using a tolerance window rather +// than strict contiguity absorbs reordered/concurrent FUSE writeback — multiple +// Write upcalls for one handle interleave (MonitorWriteAt runs before the +// per-handle write lock) and writeback flushes dirty pages slightly out of +// offset order — while still rejecting far random seeks. +const SeqTolerance = 8 << 20 // 8 MiB + +// For streaming write: keep dirty chunks in memory (NewMemChunk). +// For random write: fall back to the on-disk swap file (NewSwapFileChunk). func NewWriterPattern(chunkSize int64) *WriterPattern { return &WriterPattern{ isSequentialCounter: 0, - lastWriteStopOffset: 0, + writeFrontier: 0, chunkSize: chunkSize, } } func (rp *WriterPattern) MonitorWriteAt(offset int64, size int) { - lastOffset := atomic.SwapInt64(&rp.lastWriteStopOffset, offset+int64(size)) + // Advance the frontier to max(frontier, offset+size) and capture, in the same + // CAS loop, the pre-image this write is judged against. Reading the frontier + // inside the loop (rather than once up front) keeps `diff` below comparing + // against the freshest value even if a concurrent writeback upcall advances + // the frontier while we loop. Lock-free, consistent with the atomic counter. + end := offset + int64(size) + var frontier int64 + for { + frontier = atomic.LoadInt64(&rp.writeFrontier) + if end <= frontier || atomic.CompareAndSwapInt64(&rp.writeFrontier, frontier, end) { + break + } + } + + // near = this write starts within SeqTolerance of where writes had reached. + // Hysteresis (the ±ModeChangeLimit counter) keeps a single outlier write + // from flipping the mode and spilling a sequential stream to the swap file. + diff := offset - frontier + if diff < 0 { + diff = -diff + } counter := atomic.LoadInt64(&rp.isSequentialCounter) - if lastOffset == offset { + if diff <= SeqTolerance { if counter < ModeChangeLimit { atomic.AddInt64(&rp.isSequentialCounter, 1) } diff --git a/weed/mount/page_writer_pattern_test.go b/weed/mount/page_writer_pattern_test.go new file mode 100644 index 0000000000..5d46686cd7 --- /dev/null +++ b/weed/mount/page_writer_pattern_test.go @@ -0,0 +1,104 @@ +package mount + +import ( + "sync/atomic" + "testing" +) + +const mb = 1 << 20 + +func TestWriterPatternSequentialAndHysteresis(t *testing.T) { + wp := NewWriterPattern(2 * mb) + wp.MonitorWriteAt(0, mb) // near(0 vs frontier 0) -> +1 + wp.MonitorWriteAt(mb, mb) // +2 + wp.MonitorWriteAt(2*mb, mb) // +3 (capped) + if !wp.IsSequentialMode() { + t.Fatal("a stream from offset 0 must be sequential") + } + // a single far outlier does not flip to random (hysteresis): +3 -> +2 + wp.MonitorWriteAt(900*mb, mb) + if !wp.IsSequentialMode() { + t.Fatal("a single outlier write must not flip out of sequential mode") + } +} + +func TestWriterPatternFarFirstWriteIsRandom(t *testing.T) { + wp := NewWriterPattern(2 * mb) + wp.MonitorWriteAt(500*mb, mb) // far from frontier 0 -> -1 + if wp.IsSequentialMode() { + t.Fatal("a far first write should be random") + } +} + +func TestWriterPatternToleranceAbsorbsReorder(t *testing.T) { + wp := NewWriterPattern(2 * mb) + wp.MonitorWriteAt(0, mb) // +1, frontier 1MB + wp.MonitorWriteAt(6*mb, mb) // |6-1|=5MB <= 8MB tolerance -> near, +2, frontier 7MB + wp.MonitorWriteAt(3*mb, mb) // |3-7|=4MB <= 8MB -> near, +3 + if !wp.IsSequentialMode() { + t.Fatal("reordered writes within tolerance must stay sequential") + } +} + +func TestWriterPatternRandomRecoveryNeedsHysteresis(t *testing.T) { + wp := NewWriterPattern(2 * mb) + wp.MonitorWriteAt(100*mb, mb) // far -> -1 + wp.MonitorWriteAt(300*mb, mb) // far -> -2 + wp.MonitorWriteAt(500*mb, mb) // far -> -3 (capped), frontier 501MB + // a single near write must not immediately flip back to sequential: -3 -> -2 + wp.MonitorWriteAt(501*mb, mb) + if wp.IsSequentialMode() { + t.Fatal("one near write must not flip back from deep random mode") + } +} + +// The max-frontier (vs the old atomic.Swap of the last write's stop offset) is the +// load-bearing difference of this detector: the frontier must never move backward, +// or a backward write would lower the baseline and make later far writes look near. +func TestWriterPatternFrontierNeverRegresses(t *testing.T) { + wp := NewWriterPattern(2 * mb) + wp.MonitorWriteAt(500*mb, mb) // far first write -> -1, frontier 501MB + wp.MonitorWriteAt(0, mb) // a backward write must not pull the frontier back + if got := atomic.LoadInt64(&wp.writeFrontier); got != 501*mb { + t.Fatalf("frontier regressed to %d; the max-frontier must never move backward", got) + } + // Behavioral consequence: writes far below the preserved frontier stay random. + // Had the frontier regressed to ~1MB, this would wrongly read as sequential. + if wp.IsSequentialMode() { + t.Fatal("writes far below the preserved frontier must remain random") + } +} + +// SeqTolerance is the central new tuning knob; pin both sides of the inclusive +// boundary so an off-by-one or a '<' vs '<=' change can't slip through silently. +func TestWriterPatternToleranceBoundary(t *testing.T) { + atBoundary := NewWriterPattern(2 * mb) + atBoundary.MonitorWriteAt(SeqTolerance, 0) // diff == SeqTolerance -> near (inclusive), +1 + if !atBoundary.IsSequentialMode() { + t.Fatal("a write exactly at frontier+SeqTolerance must count as sequential") + } + pastBoundary := NewWriterPattern(2 * mb) + pastBoundary.MonitorWriteAt(SeqTolerance+1, 0) // diff == SeqTolerance+1 -> far, -1 + if pastBoundary.IsSequentialMode() { + t.Fatal("a write one byte past frontier+SeqTolerance must count as random") + } +} + +// The perf-relevant escape from random (swap-file) mode: sustained near writes must +// climb the counter back out of the negative floor and re-enter sequential (RAM) mode. +func TestWriterPatternRecoversToSequential(t *testing.T) { + wp := NewWriterPattern(2 * mb) + wp.MonitorWriteAt(100*mb, mb) // far -> -1 + wp.MonitorWriteAt(300*mb, mb) // far -> -2 + wp.MonitorWriteAt(500*mb, mb) // far -> -3 (capped), frontier 501MB + if wp.IsSequentialMode() { + t.Fatal("three far writes must be random") + } + // contiguous near writes: -3 -> -2 -> -1 -> 0 -> +1, back into sequential mode + for i := int64(0); i < 4; i++ { + wp.MonitorWriteAt(501*mb+i*mb, mb) + } + if !wp.IsSequentialMode() { + t.Fatal("sustained near writes must recover sequential mode (escape the swap-file cliff)") + } +}