diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 6c4fa0de0..2c3f588d5 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -144,6 +144,14 @@ type WFS struct { dirIdleEvict time.Duration fileIdPool *FileIdPool + // openMtimeCache maps inode -> [mtime_sec, mtime_ns] from the last Open. + // Used to decide whether to set FOPEN_KEEP_CACHE on subsequent opens. + // Bounded to openMtimeCacheMaxSize entries; when full a random entry is + // evicted. This trades a small amount of cache-miss overhead for + // predictable memory usage on mounts that touch many files. + openMtimeMu sync.Mutex + openMtimeCache map[uint64][2]int64 + // asyncFlushWg tracks pending background flush work items for writebackCache mode. // Must be waited on before unmount cleanup to prevent data loss. asyncFlushWg sync.WaitGroup @@ -221,6 +229,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { posixLocks: NewPosixLockTable(), refreshingDirs: make(map[util.FullPath]struct{}), atimeMap: make(map[uint64]time.Time, 8192), + openMtimeCache: make(map[uint64][2]int64, 8192), dirMtimeMap: make(map[uint64]time.Time, 1024), entryValidSec: 1, attrValidSec: 1, diff --git a/weed/mount/weedfs_attr.go b/weed/mount/weedfs_attr.go index d1438f970..fc0961fa1 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -74,6 +74,9 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse if size, ok := input.GetSize(); ok { glog.V(4).Infof("%v setattr set size=%v chunks=%d", path, size, len(entry.GetChunks())) + // Invalidate the open-mtime cache so the next Open does not set + // FOPEN_KEEP_CACHE with stale kernel page cache data. + wfs.invalidateOpenMtimeCache(input.NodeId) if size < filer.FileSize(entry) { // fmt.Printf("truncate %v \n", fullPath) var chunks []*filer_pb.FileChunk diff --git a/weed/mount/weedfs_file_io.go b/weed/mount/weedfs_file_io.go index cd63e6805..572a3f05a 100644 --- a/weed/mount/weedfs_file_io.go +++ b/weed/mount/weedfs_file_io.go @@ -64,7 +64,15 @@ func (wfs *WFS) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.OpenOut) if status == fuse.OK { out.Fh = uint64(fileHandle.fh) out.OpenFlags = 0 - // TODO https://github.com/libfuse/libfuse/blob/master/include/fuse_common.h#L64 + + // For read-only opens, set FOPEN_KEEP_CACHE when the file's mtime + // has not changed since the last open. This tells the kernel to + // preserve its existing page cache, avoiding redundant reads. + if in.Flags&fuse.O_ANYWRITE == 0 { + if entry := fileHandle.GetEntry(); entry != nil && entry.Attributes != nil { + wfs.applyKeepCacheFlag(in.NodeId, entry, out) + } + } } return status } @@ -94,6 +102,36 @@ func (wfs *WFS) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.OpenOut) * @param ino the inode number * @param fi file information */ +const openMtimeCacheMaxSize = 8192 + +// applyKeepCacheFlag compares the entry's mtime (seconds + nanoseconds) against +// the last-seen value and sets FOPEN_KEEP_CACHE when unchanged. +func (wfs *WFS) applyKeepCacheFlag(inode uint64, entry *LockedEntry, out *fuse.OpenOut) { + currentMtime := [2]int64{entry.Attributes.Mtime, int64(entry.Attributes.MtimeNs)} + wfs.openMtimeMu.Lock() + prev, loaded := wfs.openMtimeCache[inode] + if loaded && prev == currentMtime { + out.OpenFlags |= fuse.FOPEN_KEEP_CACHE + } else { + if len(wfs.openMtimeCache) >= openMtimeCacheMaxSize { + for k := range wfs.openMtimeCache { + delete(wfs.openMtimeCache, k) + break + } + } + wfs.openMtimeCache[inode] = currentMtime + } + wfs.openMtimeMu.Unlock() +} + +// invalidateOpenMtimeCache removes an inode's cached mtime so the next Open +// does not set FOPEN_KEEP_CACHE with stale kernel page cache data. +func (wfs *WFS) invalidateOpenMtimeCache(inode uint64) { + wfs.openMtimeMu.Lock() + delete(wfs.openMtimeCache, inode) + wfs.openMtimeMu.Unlock() +} + func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) { if in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK != 0 { wfs.posixLocks.ReleaseFlockOwner(in.NodeId, in.LockOwner) diff --git a/weed/mount/weedfs_file_io_test.go b/weed/mount/weedfs_file_io_test.go new file mode 100644 index 000000000..1881c8134 --- /dev/null +++ b/weed/mount/weedfs_file_io_test.go @@ -0,0 +1,206 @@ +package mount + +import ( + "testing" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +func newTestWFS() *WFS { + return &WFS{ + openMtimeCache: make(map[uint64][2]int64, 8192), + } +} + +func TestOpenKeepCache_FirstOpen(t *testing.T) { + // First open of a file should NOT set FOPEN_KEEP_CACHE because there + // is no previously cached mtime to compare against. + wfs := newTestWFS() + + var out fuse.OpenOut + inode := uint64(42) + + entry := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 123}, + }, + } + + wfs.applyKeepCacheFlag(inode, entry, &out) + + if out.OpenFlags&fuse.FOPEN_KEEP_CACHE != 0 { + t.Error("first open should not set FOPEN_KEEP_CACHE") + } +} + +func TestOpenKeepCache_SecondOpenSameMtime(t *testing.T) { + // Second open with an unchanged mtime SHOULD set FOPEN_KEEP_CACHE. + wfs := newTestWFS() + + inode := uint64(42) + + entry := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 123}, + }, + } + + // First open -- populate cache. + var out1 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry, &out1) + + // Second open -- mtime unchanged. + var out2 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry, &out2) + + if out2.OpenFlags&fuse.FOPEN_KEEP_CACHE == 0 { + t.Error("second open with same mtime should set FOPEN_KEEP_CACHE") + } +} + +func TestOpenKeepCache_MtimeChanged(t *testing.T) { + // If the file's mtime changes between opens, FOPEN_KEEP_CACHE must NOT + // be set so the kernel invalidates its page cache. + wfs := newTestWFS() + + inode := uint64(42) + + entry1 := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 0}, + }, + } + + // First open. + var out1 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry1, &out1) + + // File is modified externally -- mtime changes. + entry2 := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 2000, MtimeNs: 0}, + }, + } + + var out2 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry2, &out2) + + if out2.OpenFlags&fuse.FOPEN_KEEP_CACHE != 0 { + t.Error("open after mtime change should not set FOPEN_KEEP_CACHE") + } +} + +func TestOpenKeepCache_NanosecondPrecision(t *testing.T) { + // Two modifications within the same second but different nanoseconds + // must NOT reuse cached page data. + wfs := newTestWFS() + + inode := uint64(42) + + entry1 := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 100}, + }, + } + + var out1 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry1, &out1) + + // Same second, different nanosecond. + entry2 := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 200}, + }, + } + + var out2 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry2, &out2) + + if out2.OpenFlags&fuse.FOPEN_KEEP_CACHE != 0 { + t.Error("open after nanosecond-level mtime change should not set FOPEN_KEEP_CACHE") + } +} + +func TestOpenKeepCache_WriteInvalidation(t *testing.T) { + // After a write invalidates the mtime cache, the next open should NOT + // set FOPEN_KEEP_CACHE. + wfs := newTestWFS() + + inode := uint64(42) + + entry := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 0}, + }, + } + + // First open -- populate cache. + var out1 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry, &out1) + + // Simulate write invalidation. + wfs.invalidateOpenMtimeCache(inode) + + // Next open -- cache was invalidated. + var out2 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry, &out2) + + if out2.OpenFlags&fuse.FOPEN_KEEP_CACHE != 0 { + t.Error("open after write invalidation should not set FOPEN_KEEP_CACHE") + } +} + +func TestOpenKeepCache_WriteOpenSkipped(t *testing.T) { + // Write-mode opens should never evaluate FOPEN_KEEP_CACHE. + // The caller (WFS.Open) gates on O_ANYWRITE before calling + // applyKeepCacheFlag, so we verify the gate logic here. + wfs := newTestWFS() + + inode := uint64(42) + + entry := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 0}, + }, + } + + // Populate cache. + var out1 fuse.OpenOut + wfs.applyKeepCacheFlag(inode, entry, &out1) + + // Simulate write-mode open: the caller would skip applyKeepCacheFlag. + var out2 fuse.OpenOut + flags := uint32(fuse.O_ANYWRITE) + if flags&fuse.O_ANYWRITE == 0 { + wfs.applyKeepCacheFlag(inode, entry, &out2) + } + + if out2.OpenFlags&fuse.FOPEN_KEEP_CACHE != 0 { + t.Error("write open should not set FOPEN_KEEP_CACHE") + } +} + +func TestOpenKeepCache_BoundedEviction(t *testing.T) { + // Verify the cache doesn't grow beyond openMtimeCacheMaxSize. + wfs := newTestWFS() + + entry := &LockedEntry{ + Entry: &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 1000, MtimeNs: 0}, + }, + } + + for i := uint64(0); i < openMtimeCacheMaxSize+100; i++ { + var out fuse.OpenOut + wfs.applyKeepCacheFlag(i, entry, &out) + } + + wfs.openMtimeMu.Lock() + size := len(wfs.openMtimeCache) + wfs.openMtimeMu.Unlock() + + if size > openMtimeCacheMaxSize { + t.Errorf("cache size %d exceeds max %d", size, openMtimeCacheMaxSize) + } +} diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index c50702da3..8d4d8d601 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -489,6 +489,7 @@ func (wfs *WFS) truncateEntry(entryFullPath util.FullPath, entry *filer_pb.Entry } if inode, found := wfs.inodeToPath.GetInode(entryFullPath); found { + wfs.invalidateOpenMtimeCache(inode) if fh, fhFound := wfs.fhMap.FindFileHandle(inode); fhFound { fhActiveLock := fh.wfs.fhLockTable.AcquireLock("truncateEntry", fh.fh, util.ExclusiveLock) fh.ResetDirtyPages() diff --git a/weed/mount/weedfs_file_write.go b/weed/mount/weedfs_file_write.go index 63c621abe..ea64cb622 100644 --- a/weed/mount/weedfs_file_write.go +++ b/weed/mount/weedfs_file_write.go @@ -88,6 +88,9 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr fh.dirtyMetadata = true + // Invalidate the mtime cache so the next Open will not set FOPEN_KEEP_CACHE. + wfs.invalidateOpenMtimeCache(in.NodeId) + // POSIX: clear SUID/SGID bits on write by non-root users. if in.Uid != 0 { entry.Attributes.FileMode &^= 0o6000