mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-08 15:41:15 +02:00
Chris Lu
d7865909ba
* feat(mount): proactive flush of idle writable chunks Add a background goroutine that periodically scans writable chunks across all open file handles and seals those that are idle and unlikely to receive further writes, submitting them for async upload. A writable chunk is proactively flushed when it has been idle for 500ms AND meets one of: nearly full (>=90%), behind the sequential write frontier by 2+ chunks, or stale for 5+ seconds. Flushing only happens when the upload pipeline has spare capacity (< half of concurrent writer slots in use). This prevents partial chunks from accumulating until fsync/close, which is particularly beneficial for bursty or small-file workloads where chunks may never reach IsComplete(). Also fixes a latent bug in ActivityScore where MarkRead/MarkWrite used value receivers, silently discarding all mutations. * refactor(mount): reuse WriterPattern instead of duplicating sequential detection Remove the isSequential atomic from UploadPipeline. The proactive flusher now reads IsSequentialMode() from the existing WriterPattern on PageWriter and passes it as a parameter to ProactiveFlush. This avoids duplicating the sequential/random detection that WriterPattern already maintains. * fix(mount): address PR review feedback - Make ActivityScore thread-safe using atomics (CAS loop for score updates, atomic load/swap for timestamp). Previously MarkRead was called under RLock while MarkWrite held a write lock, creating a data race on the shared fields. - Fix ProactiveFlush half-capacity guard: use multiplication (uploaderCount*2 >= max) instead of floor division (max/2) which misbehaves for odd or small concurrentWriterMax values. * fix(mount): review fixes for proactive flush - Fix TOCTOU race in lastWriteChunkIndex update: use CAS loop so concurrent writers cannot regress the frontier. - Remove unused UploaderCount() getter. - Reuse the caller-provided tsNs instead of calling time.Now() again in WriteDataAt for lastWriteTsNs, eliminating a redundant syscall per write.
49 lines
1.2 KiB
Go
49 lines
1.2 KiB
Go
package mount
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
)
|
|
|
|
const (
|
|
proactiveFlushInterval = 200 * time.Millisecond
|
|
proactiveIdleThreshold = 500 * time.Millisecond
|
|
proactiveMaxHoldTime = 5 * time.Second
|
|
proactiveFillRatioNumer = 9
|
|
proactiveFillRatioDenom = 10
|
|
proactiveFrontierLag = 2
|
|
)
|
|
|
|
func (wfs *WFS) loopProactiveFlush() {
|
|
ticker := time.NewTicker(proactiveFlushInterval)
|
|
defer ticker.Stop()
|
|
|
|
glog.V(0).Infof("proactive chunk flusher started (idle=%v maxHold=%v)", proactiveIdleThreshold, proactiveMaxHoldTime)
|
|
|
|
for range ticker.C {
|
|
wfs.proactiveFlushOnce()
|
|
}
|
|
}
|
|
|
|
func (wfs *WFS) proactiveFlushOnce() {
|
|
nowNs := time.Now().UnixNano()
|
|
idleNs := proactiveIdleThreshold.Nanoseconds()
|
|
maxHoldNs := proactiveMaxHoldTime.Nanoseconds()
|
|
fillRatio := wfs.option.ChunkSizeLimit * proactiveFillRatioNumer / proactiveFillRatioDenom
|
|
|
|
var handles []*FileHandle
|
|
wfs.fhMap.RLock()
|
|
for _, fh := range wfs.fhMap.inode2fh {
|
|
if fh != nil && fh.dirtyPages != nil {
|
|
handles = append(handles, fh)
|
|
}
|
|
}
|
|
wfs.fhMap.RUnlock()
|
|
|
|
for _, fh := range handles {
|
|
isSeq := fh.dirtyPages.writerPattern.IsSequentialMode()
|
|
fh.dirtyPages.ProactiveFlush(nowNs, idleNs, maxHoldNs, fillRatio, proactiveFrontierLag, isSeq)
|
|
}
|
|
}
|