diff --git a/weed/mount/dir_sink.go b/weed/mount/dir_sink.go index 1299e4f35..de73f7dd7 100644 --- a/weed/mount/dir_sink.go +++ b/weed/mount/dir_sink.go @@ -1,6 +1,8 @@ package mount import ( + "context" + "github.com/seaweedfs/go-fuse/v2/fuse" ) @@ -39,5 +41,5 @@ func (l fuseDirEntryList) TakesLookupRef() bool { return true } // ReadDirectoryInto runs a readdir against sink. ReadDir and ReadDirPlus are // this with the kernel reply buffer as the sink. func (wfs *WFS) ReadDirectoryInto(input *fuse.ReadIn, sink DirEntrySink, isPlusMode bool) fuse.Status { - return wfs.doReadDirectory(input, sink, isPlusMode) + return wfs.doReadDirectory(context.Background(), input, sink, isPlusMode) } diff --git a/weed/mount/meta_cache/meta_cache_build_test.go b/weed/mount/meta_cache/meta_cache_build_test.go index e54f9fbe2..f2fa4e2d1 100644 --- a/weed/mount/meta_cache/meta_cache_build_test.go +++ b/weed/mount/meta_cache/meta_cache_build_test.go @@ -110,7 +110,7 @@ func TestEnsureVisitedReplaysBufferedEventsAfterSnapshot(t *testing.T) { }, } - if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if applyErr != nil { @@ -504,7 +504,7 @@ func TestEnsureVisitedPreservesLocalOnlyEntry(t *testing.T) { }}, }} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/dir")) { @@ -550,7 +550,7 @@ func TestEnsureVisitedDropsUnpinnedStaleEntry(t *testing.T) { }}, }} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/stale.txt")); err != filer_pb.ErrNotFound || entry != nil { @@ -597,7 +597,7 @@ func TestEnsureVisitedConfirmsTransientEmptyListing(t *testing.T) { }, }} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/dir")) { @@ -619,7 +619,7 @@ func TestEnsureVisitedCachesGenuinelyEmptyDirectory(t *testing.T) { } accessor := &buildFilerAccessor{client: client} - if err := EnsureVisited(mc, accessor, util.FullPath("/empty"), 0); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/empty"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/empty")) { diff --git a/weed/mount/meta_cache/meta_cache_init.go b/weed/mount/meta_cache/meta_cache_init.go index e3d8be7b9..f4bdf9b21 100644 --- a/weed/mount/meta_cache/meta_cache_init.go +++ b/weed/mount/meta_cache/meta_cache_init.go @@ -26,7 +26,7 @@ func (e *DirectoryTooLargeError) Error() string { // maxCacheableEntries is the directory size above which a build gives up, or 0 // to cache everything. -func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath, maxCacheableEntries int) error { +func EnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath, maxCacheableEntries int) error { // Collect all uncached paths from target directory up to root var uncachedPaths []util.FullPath currentPath := dirPath @@ -61,7 +61,7 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full // Fetch all uncached directories in parallel with context for cancellation // If one fetch fails, cancel the others to avoid unnecessary work - g, ctx := errgroup.WithContext(context.Background()) + g, ctx := errgroup.WithContext(ctx) for _, p := range uncachedPaths { path := p // capture for closure g.Go(func() error { @@ -91,8 +91,10 @@ const ( ) func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, path util.FullPath, maxCacheableEntries int) error { - // Use singleflight to deduplicate concurrent requests for the same path - _, err, _ := mc.visitGroup.Do(string(path), func() (interface{}, error) { + // Use singleflight to deduplicate concurrent requests for the same path. + // DoChan so a canceled request stops waiting even when another request + // owns the in-flight build. + resultCh := mc.visitGroup.DoChan(string(path), func() (interface{}, error) { // Check for cancellation before starting if ctx.Err() != nil { return nil, ctx.Err() @@ -222,7 +224,12 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl cleanupDone = true // Prevent deferred cleanup after successful publish return nil, nil }) - return err + select { + case result := <-resultCh: + return result.Err + case <-ctx.Done(): + return ctx.Err() + } } func IsHiddenSystemEntry(dir, name string) bool { diff --git a/weed/mount/meta_cache/oversized_dir_test.go b/weed/mount/meta_cache/oversized_dir_test.go index 4edb87ba0..2c3ed5048 100644 --- a/weed/mount/meta_cache/oversized_dir_test.go +++ b/weed/mount/meta_cache/oversized_dir_test.go @@ -37,7 +37,7 @@ func TestEnsureVisitedRefusesOversizedDirectory(t *testing.T) { accessor := &buildFilerAccessor{client: &buildListClient{responses: listResponses(10)}} - err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 5) + err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 5) var tooLarge *DirectoryTooLargeError if !errors.As(err, &tooLarge) { t.Fatalf("EnsureVisited = %v, want DirectoryTooLargeError", err) @@ -61,7 +61,7 @@ func TestEnsureVisitedRefusesOversizedDirectory(t *testing.T) { } // A second visit fails fast without streaming to the limit again. - if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 5); !errors.As(err, &tooLarge) { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 5); !errors.As(err, &tooLarge) { t.Fatalf("second EnsureVisited = %v, want DirectoryTooLargeError", err) } } @@ -74,7 +74,7 @@ func TestEnsureVisitedStepsOverOversizedAncestor(t *testing.T) { mc.markOversized(util.FullPath("/huge")) accessor := &buildFilerAccessor{client: &buildListClient{responses: listResponses(3)}} - if err := EnsureVisited(mc, accessor, util.FullPath("/huge/sub"), 5); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/huge/sub"), 5); err != nil { t.Fatalf("EnsureVisited under an oversized ancestor: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/huge/sub")) { @@ -92,7 +92,7 @@ func TestEnsureVisitedUnderTheLimitStillCaches(t *testing.T) { defer mc.Shutdown() accessor := &buildFilerAccessor{client: &buildListClient{responses: listResponses(5)}} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 5); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/dir"), 5); err != nil { t.Fatalf("EnsureVisited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/dir")) { @@ -123,7 +123,7 @@ func TestEnsureVisitedAncestorFoundOversizedMidVisit(t *testing.T) { "/huge/sub": listResponses(3), }}} - if err := EnsureVisited(mc, accessor, util.FullPath("/huge/sub"), 5); err != nil { + if err := EnsureVisited(context.Background(), mc, accessor, util.FullPath("/huge/sub"), 5); err != nil { t.Fatalf("EnsureVisited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/huge/sub")) { diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 1c7e2eb47..7ee15a2ee 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -502,12 +502,30 @@ func (wfs *WFS) Init(server *fuse.Server) { wfs.fuseServer = server } +// ctxFromCancel bridges a FUSE request's interrupt channel into a context so +// metadata lookups can abandon a stalled filer call when the kernel has given +// up on the operation. +func ctxFromCancel(cancel <-chan struct{}) (context.Context, context.CancelFunc) { + ctx, cancelFunc := context.WithCancel(context.Background()) + if cancel == nil { + return ctx, cancelFunc + } + go func() { + select { + case <-cancel: + cancelFunc() + case <-ctx.Done(): + } + }() + return ctx, cancelFunc +} + // maybeReadEntry resolves an inode to the entry metadata operations act on. An // open handle answers ahead of the path: unlink drops the name while the // descriptor stays valid, so an unlinked-but-open file returns its handle's // entry with an empty path instead of ENOENT. A removed directory has no // handle; its remembered entry answers the same way. -func (wfs *WFS) maybeReadEntry(inode uint64) (path util.FullPath, fh *FileHandle, entry *filer_pb.Entry, status fuse.Status) { +func (wfs *WFS) maybeReadEntry(ctx context.Context, inode uint64) (path util.FullPath, fh *FileHandle, entry *filer_pb.Entry, status fuse.Status) { var found bool if fh, found = wfs.fhMap.FindFileHandle(inode); found { path, _ = wfs.inodeToPath.GetPath(inode) @@ -525,7 +543,7 @@ func (wfs *WFS) maybeReadEntry(inode uint64) (path util.FullPath, fh *FileHandle } return } - entry, _, status = wfs.maybeLoadEntry(path) + entry, _, status = wfs.maybeLoadEntry(ctx, path) return } @@ -554,7 +572,7 @@ func (wfs *WFS) isLocalOnlyEntry(entry *filer.Entry) bool { // maybeLoadEntry returns the entry and the log position it reflects, or a zero // position when unknown. -func (wfs *WFS) maybeLoadEntry(fullpath util.FullPath) (*filer_pb.Entry, entryVersion, fuse.Status) { +func (wfs *WFS) maybeLoadEntry(ctx context.Context, fullpath util.FullPath) (*filer_pb.Entry, entryVersion, fuse.Status) { // glog.V(3).Infof("read entry cache miss %s", fullpath) _, name := fullpath.DirAndName() @@ -573,7 +591,7 @@ func (wfs *WFS) maybeLoadEntry(fullpath util.FullPath) (*filer_pb.Entry, entryVe }, entryVersion{}, fuse.OK } - entry, version, status := wfs.lookupEntry(fullpath) + entry, version, status := wfs.lookupEntry(ctx, fullpath) if status != fuse.OK { return nil, entryVersion{}, status } @@ -592,7 +610,7 @@ const expiredDirRebuildCooldown = 30 * time.Second // local entries do not leak back into lookup results. // It also returns the log position the entry reflects: the entry's stored // version, the lookup response's, or zero if unknown. -func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, fuse.Status) { +func (wfs *WFS) lookupEntry(ctx context.Context, fullpath util.FullPath) (*filer.Entry, entryVersion, fuse.Status) { dir, _ := fullpath.DirAndName() dirPath := util.FullPath(dir) @@ -605,9 +623,11 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, // flush of the target entry land first so the rebuilt cache does not // capture pre-flush metadata and bypass the wait below. if inode, found := wfs.inodeToPath.GetInode(fullpath); found { - wfs.waitForPendingAsyncFlush(inode) + if !wfs.waitForPendingAsyncFlush(ctx, inode) { + return nil, entryVersion{}, fuse.EINTR + } } - if err := wfs.ensureDirectoryVisited(dirPath); err != nil { + if err := wfs.ensureDirectoryVisited(ctx, dirPath); err != nil { // Record the attempt so the cooldown suppresses repeated rebuilds // while the listing keeps failing; once it elapses a later lookup // retries. Oversized dirs are already marked read-through. @@ -619,7 +639,7 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, } if wfs.metaCache.IsDirectoryCached(dirPath) && wfs.metaCache.IsNameFresh(fullpath) { - cachedEntry, cachedVersionTsNs, cacheErr := wfs.metaCache.FindEntry(context.Background(), fullpath) + cachedEntry, cachedVersionTsNs, cacheErr := wfs.metaCache.FindEntry(ctx, fullpath) if cacheErr != nil && cacheErr != filer_pb.ErrNotFound { glog.Errorf("lookupEntry: cache lookup for %s failed: %v", fullpath, cacheErr) return nil, entryVersion{}, fuse.EIO @@ -650,7 +670,9 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, // truncate's old size or a write's old chunks. The cache paths above are // already consistent and must not pay this wait. if inode, found := wfs.inodeToPath.GetInode(fullpath); found { - wfs.waitForPendingAsyncFlush(inode) + if !wfs.waitForPendingAsyncFlush(ctx, inode) { + return nil, entryVersion{}, fuse.EINTR + } } // Directory not cached - fetch directly from filer without caching the entire directory. @@ -658,7 +680,7 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, var entry *filer_pb.Entry var lookupVersion entryVersion lookupDir, lookupName := fullpath.DirAndName() - err := wfs.withFilerClient(context.Background(), false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { + err := wfs.withFilerClient(ctx, false, wfs.filerRPCWait(), func(ctx context.Context, client filer_pb.SeaweedFilerClient) error { resp, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ Directory: lookupDir, Name: lookupName, @@ -671,6 +693,9 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, return nil }) if err != nil { + if ctx.Err() != nil { + return nil, entryVersion{}, fuse.EINTR + } if err == filer_pb.ErrNotFound { // The entry may exist in the local store from a deferred create // (deferFilerCreate=true) that hasn't been flushed yet. Only trust @@ -689,7 +714,7 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, wfs.pendingAsyncFlushMu.Unlock() if hasDirtyHandle || hasPendingFlush { - if localEntry, localVersionTsNs, localErr := wfs.metaCache.FindEntry(context.Background(), fullpath); localErr == nil && localEntry != nil { + if localEntry, localVersionTsNs, localErr := wfs.metaCache.FindEntry(ctx, fullpath); localErr == nil && localEntry != nil { glog.V(4).Infof("lookupEntry found deferred entry in local cache %s", fullpath) return localEntry, entryVersion{tsNs: localVersionTsNs}, fuse.OK } @@ -723,7 +748,7 @@ func (wfs *WFS) lookupEntry(fullpath util.FullPath) (*filer.Entry, entryVersion, // up without -v=4 — and include layer-by-layer state so the // next failed run pinpoints which layer dropped the entry. localPresent := false - if localEntry, _, localErr := wfs.metaCache.FindEntry(context.Background(), fullpath); localErr == nil && localEntry != nil { + if localEntry, _, localErr := wfs.metaCache.FindEntry(ctx, fullpath); localErr == nil && localEntry != nil { localPresent = true } glog.Warningf("lookupEntry: filer ErrNotFound for tracked path %s (inode=%d dirtyHandle=%v pendingFlush=%v localCache=%v dirCached=%v) — possible coherence bug", diff --git a/weed/mount/weedfs_access.go b/weed/mount/weedfs_access.go index 27fa83132..8eb402b69 100644 --- a/weed/mount/weedfs_access.go +++ b/weed/mount/weedfs_access.go @@ -82,7 +82,9 @@ func clearSupplementaryGroupCache() { * This method is not called under Linux kernel versions 2.4.x */ func (wfs *WFS) Access(cancel <-chan struct{}, input *fuse.AccessIn) (code fuse.Status) { - _, _, entry, code := wfs.maybeReadEntry(input.NodeId) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + _, _, entry, code := wfs.maybeReadEntry(ctx, input.NodeId) if code != fuse.OK { return code } diff --git a/weed/mount/weedfs_attr.go b/weed/mount/weedfs_attr.go index 39c34e35c..b592ad921 100644 --- a/weed/mount/weedfs_attr.go +++ b/weed/mount/weedfs_attr.go @@ -26,7 +26,9 @@ func (wfs *WFS) GetAttr(cancel <-chan struct{}, input *fuse.GetAttrIn, out *fuse } inode := input.NodeId - path, fh, entry, status := wfs.maybeReadEntry(inode) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + path, fh, entry, status := wfs.maybeReadEntry(ctx, inode) if status != fuse.OK { return status } @@ -65,7 +67,9 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse return fuse.Status(syscall.ENOSPC) } - path, fh, entry, status := wfs.maybeReadEntry(input.NodeId) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + path, fh, entry, status := wfs.maybeReadEntry(ctx, input.NodeId) if status != fuse.OK || entry == nil { return status } @@ -336,7 +340,7 @@ func (wfs *WFS) touchDirMtimeCtimeBest(dirPath util.FullPath) { // touchDirMtimeCtime updates a directory's mtime and ctime on the filer. // POSIX requires this when entries are created or removed in the directory. func (wfs *WFS) touchDirMtimeCtime(dirPath util.FullPath) { - dirEntry, _, code := wfs.maybeLoadEntry(dirPath) + dirEntry, _, code := wfs.maybeLoadEntry(context.Background(), dirPath) if code != fuse.OK || dirEntry == nil || dirEntry.Attributes == nil { return } diff --git a/weed/mount/weedfs_dir_lookup.go b/weed/mount/weedfs_dir_lookup.go index 24e39f684..60bd56e90 100644 --- a/weed/mount/weedfs_dir_lookup.go +++ b/weed/mount/weedfs_dir_lookup.go @@ -27,7 +27,9 @@ func (wfs *WFS) Lookup(cancel <-chan struct{}, header *fuse.InHeader, name strin fullFilePath := dirPath.Child(name) // Use shared lookup logic that checks cache first, then filer if needed - localEntry, _, status := wfs.lookupEntry(fullFilePath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + localEntry, _, status := wfs.lookupEntry(ctx, fullFilePath) if status != fuse.OK { return status } diff --git a/weed/mount/weedfs_dir_lookup_cache_test.go b/weed/mount/weedfs_dir_lookup_cache_test.go index 6df1128c6..f1f3635cd 100644 --- a/weed/mount/weedfs_dir_lookup_cache_test.go +++ b/weed/mount/weedfs_dir_lookup_cache_test.go @@ -158,7 +158,7 @@ func TestLookupRebuildsExpiredDirectoryCache(t *testing.T) { } startFakeFiler(t, wfs, fake) - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dir, 0); err != nil { + if err := meta_cache.EnsureVisited(context.Background(), wfs.metaCache, wfs, dir, 0); err != nil { t.Fatalf("EnsureVisited: %v", err) } if !wfs.inodeToPath.IsChildrenCached(dir) { @@ -174,7 +174,7 @@ func TestLookupRebuildsExpiredDirectoryCache(t *testing.T) { lookupCallsBefore := fake.lookupCalls.Load() for _, name := range []string{"a", "b", "c"} { - entry, _, status := wfs.lookupEntry(dir.Child(name)) + entry, _, status := wfs.lookupEntry(context.Background(), dir.Child(name)) if status != fuse.OK { t.Fatalf("lookupEntry %s: %v", name, status) } @@ -211,7 +211,7 @@ func TestLookupRebuildDeduplicatesConcurrentRebuilds(t *testing.T) { } startFakeFiler(t, wfs, fake) - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dir, 0); err != nil { + if err := meta_cache.EnsureVisited(context.Background(), wfs.metaCache, wfs, dir, 0); err != nil { t.Fatalf("EnsureVisited: %v", err) } expireDirCache(t, wfs.inodeToPath, dir) @@ -228,7 +228,7 @@ func TestLookupRebuildDeduplicatesConcurrentRebuilds(t *testing.T) { wg.Add(1) go func(i int, name string) { defer wg.Done() - if _, _, status := wfs.lookupEntry(dir.Child(name)); status != fuse.OK { + if _, _, status := wfs.lookupEntry(context.Background(), dir.Child(name)); status != fuse.OK { errs[i] = fmt.Errorf("lookupEntry %s: %v", name, status) } }(i, name) @@ -271,7 +271,7 @@ func TestLookupRebuildCooldownAfterFailure(t *testing.T) { } startFakeFiler(t, wfs, fake) - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dir, 0); err != nil { + if err := meta_cache.EnsureVisited(context.Background(), wfs.metaCache, wfs, dir, 0); err != nil { t.Fatalf("EnsureVisited: %v", err) } expireDirCache(t, wfs.inodeToPath, dir) @@ -280,7 +280,7 @@ func TestLookupRebuildCooldownAfterFailure(t *testing.T) { fake.listErr = errors.New("persistent listing failure") listCallsBefore := fake.listCalls.Load() lookupCallsBefore := fake.lookupCalls.Load() - if entry, _, status := wfs.lookupEntry(dir.Child("a")); status != fuse.OK || entry == nil || entry.Name() != "a" { + if entry, _, status := wfs.lookupEntry(context.Background(), dir.Child("a")); status != fuse.OK || entry == nil || entry.Name() != "a" { t.Fatalf("lookupEntry a after failed rebuild: status=%v entry=%v", status, entry) } if got := fake.lookupCalls.Load() - lookupCallsBefore; got != 1 { @@ -293,7 +293,7 @@ func TestLookupRebuildCooldownAfterFailure(t *testing.T) { // Immediate retry: cooldown suppresses the rebuild, falls through again. listCallsBefore = fake.listCalls.Load() lookupCallsBefore = fake.lookupCalls.Load() - if _, _, status := wfs.lookupEntry(dir.Child("a")); status != fuse.OK { + if _, _, status := wfs.lookupEntry(context.Background(), dir.Child("a")); status != fuse.OK { t.Fatalf("second lookupEntry a: %v", status) } if got := fake.listCalls.Load() - listCallsBefore; got != 0 { @@ -308,7 +308,7 @@ func TestLookupRebuildCooldownAfterFailure(t *testing.T) { advanceRebuildAttempt(t, wfs.inodeToPath, dir, expiredDirRebuildCooldown+time.Second) listCallsBefore = fake.listCalls.Load() lookupCallsBefore = fake.lookupCalls.Load() - if _, _, status := wfs.lookupEntry(dir.Child("a")); status != fuse.OK { + if _, _, status := wfs.lookupEntry(context.Background(), dir.Child("a")); status != fuse.OK { t.Fatalf("third lookupEntry a: %v", status) } if got := fake.listCalls.Load() - listCallsBefore; got != 1 { diff --git a/weed/mount/weedfs_dir_mkrm.go b/weed/mount/weedfs_dir_mkrm.go index 755fbf444..61317b136 100644 --- a/weed/mount/weedfs_dir_mkrm.go +++ b/weed/mount/weedfs_dir_mkrm.go @@ -156,13 +156,20 @@ func (wfs *WFS) Rmdir(cancel <-chan struct{}, header *fuse.InHeader, name string } entryFullPath := dirFullPath.Child(name) - targetEntry, _, targetCode := wfs.maybeLoadEntry(entryFullPath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + targetEntry, _, targetCode := wfs.maybeLoadEntry(ctx, entryFullPath) + if targetCode == fuse.EINTR { + return targetCode + } if targetCode != fuse.OK { targetEntry = nil } // POSIX: enforce sticky bit on the parent directory. - if dirEntry, _, dirCode := wfs.maybeLoadEntry(dirFullPath); dirCode == fuse.OK && dirEntry != nil && dirEntry.Attributes != nil { + if dirEntry, _, dirCode := wfs.maybeLoadEntry(ctx, dirFullPath); dirCode != fuse.OK { + return dirCode + } else if dirEntry != nil && dirEntry.Attributes != nil { targetUid := uint32(0) if targetEntry != nil && targetEntry.Attributes != nil { targetUid = targetEntry.Attributes.Uid diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index 57186e08c..6e7a39159 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/weed/mount/weedfs_dir_read.go @@ -176,14 +176,20 @@ func (wfs *WFS) FsyncDir(cancel <-chan struct{}, input *fuse.FsyncIn) (code fuse * '1'. */ func (wfs *WFS) ReadDir(cancel <-chan struct{}, input *fuse.ReadIn, out *fuse.DirEntryList) (code fuse.Status) { - return wfs.doReadDirectory(input, fuseDirEntryList{out}, false) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + return wfs.doReadDirectory(ctx, input, fuseDirEntryList{out}, false) } func (wfs *WFS) ReadDirPlus(cancel <-chan struct{}, input *fuse.ReadIn, out *fuse.DirEntryList) (code fuse.Status) { - return wfs.doReadDirectory(input, fuseDirEntryList{out}, true) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + return wfs.doReadDirectory(ctx, input, fuseDirEntryList{out}, true) } -func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode bool) fuse.Status { +func (wfs *WFS) doReadDirectory(ctx context.Context, input *fuse.ReadIn, out DirEntrySink, isPlusMode bool) fuse.Status { + listingCtx := filer_pb.WithChunksOmitted(ctx) + // Get the directory handle and lock it for the duration of this operation. // This serializes concurrent readdir calls on the same handle, fixing the // race condition that caused hangs with NFS-Ganesha. @@ -282,7 +288,7 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode dh.dropConsumed(input.Offset) if wfs.inodeToPath.ShouldReadDirectoryDirect(dirPath) { - return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) } // Read from cache first, then load next batch if needed @@ -292,32 +298,38 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset { skipCount := int64(input.Offset - dh.entryStreamOffset) - if err := wfs.ensureDirectoryVisited(dirPath); err != nil { + if err := wfs.ensureDirectoryVisited(ctx, dirPath); err != nil { var tooLarge *meta_cache.DirectoryTooLargeError if errors.As(err, &tooLarge) { - return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) + } + if ctx.Err() != nil { + return fuse.EINTR } glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) return fuse.EIO } // Serve the maintained-but-unverified cache if the filer is unreachable. - if err := meta_cache.EnsureListingFresh(context.Background(), wfs.metaCache, wfs, dirPath, ""); err != nil { + if err := meta_cache.EnsureListingFresh(ctx, wfs.metaCache, wfs, dirPath, ""); err != nil { if errors.Is(err, meta_cache.ErrRefreshRangeTooLarge) { // re-tile with a full rebuild; serve this request direct wfs.purgeDirectoryCache(dirPath) - return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) } glog.V(1).Infof("refresh %s sections: %v", dirPath, err) } // Load entries from beginning to fill cache up to the requested offset - storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) { + storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(listingCtx, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) { dh.entryStream = append(dh.entryStream, entry) return true, nil }) dh.lastListedName = storeLastName if loadErr != nil { + if ctx.Err() != nil { + return fuse.EINTR + } glog.Errorf("list meta cache: %v", loadErr) return fuse.EIO } @@ -349,12 +361,15 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode } // Cache exhausted, load next batch - if err := wfs.ensureDirectoryVisited(dirPath); err != nil { + if err := wfs.ensureDirectoryVisited(ctx, dirPath); err != nil { var tooLarge *meta_cache.DirectoryTooLargeError if errors.As(err, &tooLarge) { // The direct path keeps the same pagination state on dh, so it // carries on from wherever the cached walk reached. - return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) + } + if ctx.Err() != nil { + return fuse.EINTR } glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) return fuse.EIO @@ -369,17 +384,17 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode } // Serve the maintained-but-unverified cache if the filer is unreachable. - if err := meta_cache.EnsureListingFresh(context.Background(), wfs.metaCache, wfs, dirPath, lastEntryName); err != nil { + if err := meta_cache.EnsureListingFresh(ctx, wfs.metaCache, wfs, dirPath, lastEntryName); err != nil { if errors.Is(err, meta_cache.ErrRefreshRangeTooLarge) { // re-tile with a full rebuild; serve this request direct wfs.purgeDirectoryCache(dirPath) - return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + return wfs.readDirectoryDirect(ctx, input, out, dh, dirPath, processEachEntryFn) } glog.V(1).Infof("refresh %s sections: %v", dirPath, err) } bufferFull := false - storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) { + storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(listingCtx, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) { currentIndex := int64(len(dh.entryStream)) dh.entryStream = append(dh.entryStream, entry) if !processEachEntryFn(entry, currentIndex) { @@ -389,6 +404,9 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode return true, nil }) if loadErr != nil { + if ctx.Err() != nil { + return fuse.EINTR + } glog.Errorf("list meta cache: %v", loadErr) return fuse.EIO } @@ -408,8 +426,8 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode // ensureDirectoryVisited pulls the directory into the local cache, unless it is // too large to cache: then the directory is marked read-through, so later // listings go straight to the filer without re-asking. -func (wfs *WFS) ensureDirectoryVisited(dirPath util.FullPath) error { - err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath, wfs.option.CacheDirMaxEntries) +func (wfs *WFS) ensureDirectoryVisited(ctx context.Context, dirPath util.FullPath) error { + err := meta_cache.EnsureVisited(ctx, wfs.metaCache, wfs, dirPath, wfs.option.CacheDirMaxEntries) var tooLarge *meta_cache.DirectoryTooLargeError if errors.As(err, &tooLarge) { wfs.inodeToPath.MarkDirectoryReadThrough(dirPath, time.Now()) @@ -417,14 +435,18 @@ func (wfs *WFS) ensureDirectoryVisited(dirPath util.FullPath) error { return err } -func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *DirectoryHandle, dirPath util.FullPath, processEachEntryFn func(entry *filer.Entry, index int64) bool) fuse.Status { +func (wfs *WFS) readDirectoryDirect(ctx context.Context, input *fuse.ReadIn, out DirEntrySink, dh *DirectoryHandle, dirPath util.FullPath, processEachEntryFn func(entry *filer.Entry, index int64) bool) fuse.Status { + listingCtx := filer_pb.WithChunksOmitted(ctx) var lastEntryName string if input.Offset >= dh.entryStreamOffset { if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset { skipCount := uint32(input.Offset-dh.entryStreamOffset) + batchSize - entries, snapshotTs, err := loadDirectoryEntriesDirect(readdirContext, wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) + entries, snapshotTs, err := loadDirectoryEntriesDirect(listingCtx, wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) if err != nil { + if ctx.Err() != nil { + return fuse.EINTR + } glog.Errorf("list filer directory: %v", err) return fuse.EIO } @@ -457,8 +479,11 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di } } - entries, snapshotTs, err := loadDirectoryEntriesDirect(readdirContext, wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) + entries, snapshotTs, err := loadDirectoryEntriesDirect(listingCtx, wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) if err != nil { + if ctx.Err() != nil { + return fuse.EINTR + } glog.Errorf("list filer directory: %v", err) return fuse.EIO } diff --git a/weed/mount/weedfs_dir_read_bench_test.go b/weed/mount/weedfs_dir_read_bench_test.go index a78da820d..e8c48ad83 100644 --- a/weed/mount/weedfs_dir_read_bench_test.go +++ b/weed/mount/weedfs_dir_read_bench_test.go @@ -163,7 +163,7 @@ func walkOnce(tb testing.TB, wfs *WFS, dirInode uint64, sink *benchSink, forgets offset := uint64(0) for { sink.reset() - status := wfs.doReadDirectory(&fuse.ReadIn{ + status := wfs.doReadDirectory(context.Background(), &fuse.ReadIn{ InHeader: fuse.InHeader{NodeId: dirInode}, Fh: uint64(dhid), Offset: offset, diff --git a/weed/mount/weedfs_dir_read_pagination_test.go b/weed/mount/weedfs_dir_read_pagination_test.go index ba38b3db1..63b979c18 100644 --- a/weed/mount/weedfs_dir_read_pagination_test.go +++ b/weed/mount/weedfs_dir_read_pagination_test.go @@ -111,7 +111,7 @@ func TestReadDirResumePastShrunkDirectory(t *testing.T) { defer wfs.ReleaseDirectoryHandle(dhid) sink := &pagingSink{limit: 4096} - status := wfs.doReadDirectory(&fuse.ReadIn{ + status := wfs.doReadDirectory(context.Background(), &fuse.ReadIn{ InHeader: fuse.InHeader{NodeId: dirInode}, Fh: uint64(dhid), Offset: 152, // past the end of a directory that now holds 100 @@ -158,7 +158,7 @@ func TestReadDirWithExpiredEntries(t *testing.T) { var offset uint64 for round := 0; round < 20; round++ { sink.round = 0 - status := wfs.doReadDirectory(&fuse.ReadIn{ + status := wfs.doReadDirectory(context.Background(), &fuse.ReadIn{ InHeader: fuse.InHeader{NodeId: dirInode}, Fh: uint64(dhid), Offset: offset, @@ -209,7 +209,7 @@ func TestReadDirTrimsConsumedEntries(t *testing.T) { for round := 0; round < 500; round++ { sink.round = 0 before := len(sink.names) - status := wfs.doReadDirectory(&fuse.ReadIn{ + status := wfs.doReadDirectory(context.Background(), &fuse.ReadIn{ InHeader: fuse.InHeader{NodeId: dirInode}, Fh: uint64(dhid), Offset: offset, @@ -301,7 +301,7 @@ func TestReadDirDirectTrimsConsumedEntries(t *testing.T) { for round := 0; round < 500; round++ { sink.round = 0 before := len(sink.names) - status := wfs.doReadDirectory(&fuse.ReadIn{ + status := wfs.doReadDirectory(context.Background(), &fuse.ReadIn{ InHeader: fuse.InHeader{NodeId: dirInode}, Fh: uint64(dhid), Offset: offset, diff --git a/weed/mount/weedfs_file_io.go b/weed/mount/weedfs_file_io.go index b40a489a4..bee1877fb 100644 --- a/weed/mount/weedfs_file_io.go +++ b/weed/mount/weedfs_file_io.go @@ -65,7 +65,9 @@ import ( */ func (wfs *WFS) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.OpenOut) (status fuse.Status) { var fileHandle *FileHandle - fileHandle, status = wfs.AcquireHandle(in.NodeId, in.Flags, in.Uid, in.Gid) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + fileHandle, status = wfs.AcquireHandle(ctx, in.NodeId, in.Flags, in.Uid, in.Gid) if status == fuse.OK { out.Fh = uint64(fileHandle.fh) out.OpenFlags = 0 diff --git a/weed/mount/weedfs_file_mkrm.go b/weed/mount/weedfs_file_mkrm.go index 0e7e02db1..62aa7116e 100644 --- a/weed/mount/weedfs_file_mkrm.go +++ b/weed/mount/weedfs_file_mkrm.go @@ -41,7 +41,9 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o entryFullPath := dirFullPath.Child(name) var inode uint64 - newEntry, _, code := wfs.maybeLoadEntry(entryFullPath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + newEntry, _, code := wfs.maybeLoadEntry(ctx, entryFullPath) if code == fuse.OK { if newEntry == nil || newEntry.Attributes == nil { return fuse.EIO @@ -52,7 +54,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o return fuse.Status(syscall.EEXIST) } inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, len(newEntry.HardLinkId) > 0, newEntry.Attributes.Inode, true) - fileHandle, status := wfs.AcquireHandle(inode, in.Flags, in.Uid, in.Gid) + fileHandle, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid) if status != fuse.OK { return status } @@ -86,14 +88,14 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o } } - inode, newEntry, code = wfs.createRegularFile(dirFullPath, name, in.Mode, in.Uid, in.Gid, 0, !wfs.option.EagerFilerCreate, true) + inode, newEntry, code = wfs.createRegularFile(ctx, dirFullPath, name, in.Mode, in.Uid, in.Gid, 0, !wfs.option.EagerFilerCreate, true) if code == fuse.Status(syscall.EEXIST) && in.Flags&syscall.O_EXCL == 0 { // Race: another process created the file between our check and create. // Reopen the winner's entry; AcquireHandle takes its own lock. if dlmLock != nil { dlmLock.Stop() } - newEntry, _, code = wfs.maybeLoadEntry(entryFullPath) + newEntry, _, code = wfs.maybeLoadEntry(ctx, entryFullPath) if code != fuse.OK { return code } @@ -101,7 +103,7 @@ func (wfs *WFS) Create(cancel <-chan struct{}, in *fuse.CreateIn, name string, o return fuse.EIO } inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, len(newEntry.HardLinkId) > 0, newEntry.Attributes.Inode, true) - fileHandle, status := wfs.AcquireHandle(inode, in.Flags, in.Uid, in.Gid) + fileHandle, status := wfs.AcquireHandle(ctx, inode, in.Flags, in.Uid, in.Gid) if status != fuse.OK { return status } @@ -177,7 +179,9 @@ func (wfs *WFS) Mknod(cancel <-chan struct{}, in *fuse.MknodIn, name string, out return } - inode, newEntry, code := wfs.createRegularFile(dirFullPath, name, in.Mode, in.Uid, in.Gid, in.Rdev, false, false) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + inode, newEntry, code := wfs.createRegularFile(ctx, dirFullPath, name, in.Mode, in.Uid, in.Gid, in.Rdev, false, false) if code != fuse.OK { return code } @@ -207,7 +211,9 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin } entryFullPath := dirFullPath.Child(name) - entry, _, code := wfs.maybeLoadEntry(entryFullPath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + entry, _, code := wfs.maybeLoadEntry(ctx, entryFullPath) if code != fuse.OK { if code == fuse.ENOENT { return fuse.OK @@ -236,7 +242,7 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin // maybeLoadEntry path above. Do not fall back to the stale // pre-lock snapshot: proceeding with its HardLinkCounter would // reintroduce the stale-base update the lock is meant to prevent. - fresh, _, freshCode := wfs.maybeLoadEntry(entryFullPath) + fresh, _, freshCode := wfs.maybeLoadEntry(ctx, entryFullPath) if freshCode == fuse.ENOENT { return fuse.OK } @@ -247,7 +253,9 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin } // POSIX: enforce sticky bit on the parent directory. - if dirEntry, _, dirCode := wfs.maybeLoadEntry(dirFullPath); dirCode == fuse.OK && dirEntry != nil && dirEntry.Attributes != nil { + if dirEntry, _, dirCode := wfs.maybeLoadEntry(ctx, dirFullPath); dirCode != fuse.OK { + return dirCode + } else if dirEntry != nil && dirEntry.Attributes != nil { targetUid := uint32(0) if entry != nil && entry.Attributes != nil { targetUid = entry.Attributes.Uid @@ -266,11 +274,11 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin // will remove it again. if inode, found := wfs.inodeToPath.GetInode(entryFullPath); found { wfs.markHandleDeleted(inode) - wfs.waitForPendingAsyncFlush(inode) + wfs.waitForPendingAsyncFlush(context.Background(), inode) } else if entry != nil && entry.Attributes != nil && entry.Attributes.Inode != 0 { inodeFromEntry := entry.Attributes.Inode wfs.markHandleDeleted(inodeFromEntry) - wfs.waitForPendingAsyncFlush(inodeFromEntry) + wfs.waitForPendingAsyncFlush(context.Background(), inodeFromEntry) } // first, ensure the filer store can correctly delete @@ -337,7 +345,7 @@ func (wfs *WFS) Unlink(cancel <-chan struct{}, header *fuse.InHeader, name strin } -func (wfs *WFS) createRegularFile(dirFullPath util.FullPath, name string, mode uint32, uid, gid, rdev uint32, deferFilerCreate bool, skipExistenceCheck bool) (inode uint64, newEntry *filer_pb.Entry, code fuse.Status) { +func (wfs *WFS) createRegularFile(ctx context.Context, dirFullPath util.FullPath, name string, mode uint32, uid, gid, rdev uint32, deferFilerCreate bool, skipExistenceCheck bool) (inode uint64, newEntry *filer_pb.Entry, code fuse.Status) { if wfs.IsOverQuotaWithUncommitted() { return 0, nil, fuse.Status(syscall.ENOSPC) } @@ -346,7 +354,7 @@ func (wfs *WFS) createRegularFile(dirFullPath util.FullPath, name string, mode u // SkipCheckParentDirectory, so this is the only parent check). With // default_permissions the kernel already verified write+search on it before // Create/Mknod, so skip only the mode-bit check and its group lookup. - parentEntry, _, parentStatus := wfs.maybeLoadEntry(dirFullPath) + parentEntry, _, parentStatus := wfs.maybeLoadEntry(ctx, dirFullPath) if parentStatus != fuse.OK { return 0, nil, parentStatus } @@ -367,7 +375,7 @@ func (wfs *WFS) createRegularFile(dirFullPath util.FullPath, name string, mode u entryFullPath := dirFullPath.Child(name) if !skipExistenceCheck { - if _, _, status := wfs.maybeLoadEntry(entryFullPath); status == fuse.OK { + if _, _, status := wfs.maybeLoadEntry(ctx, entryFullPath); status == fuse.OK { return 0, nil, fuse.Status(syscall.EEXIST) } else if status != fuse.ENOENT { return 0, nil, status diff --git a/weed/mount/weedfs_file_mkrm_test.go b/weed/mount/weedfs_file_mkrm_test.go index 6a8b34312..6ed637f60 100644 --- a/weed/mount/weedfs_file_mkrm_test.go +++ b/weed/mount/weedfs_file_mkrm_test.go @@ -528,7 +528,7 @@ func TestAcquireHandleHonorsDefaultPermissions(t *testing.T) { } inode := wfs.inodeToPath.Lookup(util.FullPath("/secret.txt"), entry.Attributes.Crtime, false, false, entry.Attributes.Inode, true) - fh, status := wfs.AcquireHandle(inode, syscall.O_RDONLY, 999, 999) + fh, status := wfs.AcquireHandle(context.Background(), inode, syscall.O_RDONLY, 999, 999) if status != tc.want { t.Fatalf("AcquireHandle status = %v, want %v", status, tc.want) } @@ -549,7 +549,7 @@ func TestCreateRegularFileValidatesParentUnderDefaultPermissions(t *testing.T) { wfs, _ := newCreateTestWFS(t) wfs.option.DefaultPermissions = true - if _, _, code := wfs.createRegularFile(util.FullPath("/ghost"), "f.txt", 0o644, 99, 100, 0, false, false); code == fuse.OK { + if _, _, code := wfs.createRegularFile(context.Background(), util.FullPath("/ghost"), "f.txt", 0o644, 99, 100, 0, false, false); code == fuse.OK { t.Fatal("createRegularFile returned OK for a missing parent directory") } } diff --git a/weed/mount/weedfs_file_read.go b/weed/mount/weedfs_file_read.go index 3044c3a76..33f5b6bb0 100644 --- a/weed/mount/weedfs_file_read.go +++ b/weed/mount/weedfs_file_read.go @@ -47,17 +47,8 @@ func (wfs *WFS) Read(cancel <-chan struct{}, in *fuse.ReadIn, buff []byte) (fuse fhActiveLock := fh.wfs.fhLockTable.AcquireLock("Read", fh.fh, util.SharedLock) defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) - // Create a context that will be cancelled when the cancel channel receives a signal - ctx, cancelFunc := context.WithCancel(context.Background()) - defer cancelFunc() // Ensure cleanup - - go func() { - select { - case <-cancel: - cancelFunc() - case <-ctx.Done(): - } - }() + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() offset := int64(in.Offset) totalRead, err := readDataByFileHandleWithContext(ctx, buff, fh, offset) diff --git a/weed/mount/weedfs_filehandle.go b/weed/mount/weedfs_filehandle.go index d7f16b76e..c5d25e1cb 100644 --- a/weed/mount/weedfs_filehandle.go +++ b/weed/mount/weedfs_filehandle.go @@ -1,6 +1,7 @@ package mount import ( + "context" "fmt" "syscall" @@ -10,12 +11,14 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) -func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle *FileHandle, status fuse.Status) { +func (wfs *WFS) AcquireHandle(ctx context.Context, inode uint64, flags, uid, gid uint32) (fileHandle *FileHandle, status fuse.Status) { // If there is an in-flight async flush for this inode, wait for it to // complete before reopening. Otherwise the new handle would be built // from pre-close filer metadata and its next flush could overwrite the // data that was just written asynchronously. - wfs.waitForPendingAsyncFlush(inode) + if !wfs.waitForPendingAsyncFlush(ctx, inode) { + return nil, fuse.EINTR + } path, pathStatus := wfs.inodeToPath.GetPath(inode) if pathStatus != fuse.OK { @@ -38,7 +41,7 @@ func (wfs *WFS) AcquireHandle(inode uint64, flags, uid, gid uint32) (fileHandle }) entryVersionTsNs = entryVersion{tsNs: existingFh.entryVersionTsNs.Load(), signature: existingFh.entryVersionSignature.Load()} } else { - entry, entryVersionTsNs, status = wfs.maybeLoadEntry(path) + entry, entryVersionTsNs, status = wfs.maybeLoadEntry(ctx, path) if status != fuse.OK { return nil, status } @@ -125,14 +128,20 @@ func (wfs *WFS) ReleaseHandle(handleId FileHandleId) { // waitForPendingAsyncFlush blocks until any in-flight async flush for // the given inode completes. Called from AcquireHandle before building // new handle state, so the filer metadata reflects the flushed data. -func (wfs *WFS) waitForPendingAsyncFlush(inode uint64) { +func (wfs *WFS) waitForPendingAsyncFlush(ctx context.Context, inode uint64) bool { wfs.pendingAsyncFlushMu.Lock() done, found := wfs.pendingAsyncFlush[inode] wfs.pendingAsyncFlushMu.Unlock() - if found { - glog.V(3).Infof("waitForPendingAsyncFlush: waiting for inode %d", inode) - <-done + if !found { + return true + } + glog.V(3).Infof("waitForPendingAsyncFlush: waiting for inode %d", inode) + select { + case <-done: + return true + case <-ctx.Done(): + return false } } diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index f53d23aee..d65202368 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -499,7 +499,7 @@ func TestQueuedEventDoesNotRollBackHandleOpenedAfterEnqueue(t *testing.T) { // Open now: the lookup reaches the filer, which serves the newer // size-200 state versioned at log position 2000. inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("AcquireHandle status = %v, want OK", status) } @@ -563,7 +563,7 @@ func TestAbortedBuildEventDoesNotRollBackLaterOpen(t *testing.T) { } inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("AcquireHandle status = %v, want OK", status) } @@ -626,7 +626,7 @@ func TestEventDuringOpenLookupIsFenced(t *testing.T) { } opened := make(chan openResult, 1) go func() { - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) opened <- openResult{fh, status} }() @@ -755,7 +755,7 @@ func TestLocalAckDoesNotFenceUnrelatedDelayedEvents(t *testing.T) { // Open the file: the cached-store read must not claim position 2000 — // the local ack versioned only its own path. inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("AcquireHandle status = %v, want OK", status) } @@ -794,14 +794,14 @@ func TestSlowerConcurrentOpenDoesNotOverwriteNewerHandle(t *testing.T) { } slow := make(chan openResult, 1) go func() { - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) slow <- openResult{fh, status} }() <-fake.lookupStarted // The faster opener completes with newer state while the slow lookup is // still in flight. - fastFh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fastFh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("fast AcquireHandle status = %v, want OK", status) } @@ -925,12 +925,12 @@ func TestSlowerOpenRejectedWhenHandleDirtyOrLookupUnversioned(t *testing.T) { } slow := make(chan openResult, 1) go func() { - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) slow <- openResult{fh, status} }() <-fake.lookupStarted - fastFh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fastFh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("fast AcquireHandle status = %v, want OK", status) } @@ -969,12 +969,12 @@ func TestUnversionedSlowerOpenDoesNotOverwrite(t *testing.T) { } slow := make(chan openResult, 1) go func() { - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) slow <- openResult{fh, status} }() <-fake.lookupStarted - fastFh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fastFh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("fast AcquireHandle status = %v, want OK", status) } @@ -1405,7 +1405,7 @@ func TestEmptyListingTrailerSnapshotSetsAbsenceFloor(t *testing.T) { startFakeFiler(t, wfs, &fakeFilerServer{listSnapshotTrailerTsNs: 4000}) wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false) - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, util.FullPath("/dir"), 0); err != nil { + if err := meta_cache.EnsureVisited(context.Background(), wfs.metaCache, wfs, util.FullPath("/dir"), 0); err != nil { t.Fatalf("EnsureVisited: %v", err) } @@ -1820,7 +1820,7 @@ func TestEventFromAnotherFilerIsNotFencedByForeignClock(t *testing.T) { startFakeFiler(t, wfs, &fakeFilerServer{lookupSize: 200, lookupLogTsNs: 5000, lookupSignature: 11}) inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) - fh, status := wfs.AcquireHandle(inode, 0, 0, 0) + fh, status := wfs.AcquireHandle(context.Background(), inode, 0, 0, 0) if status != fuse.OK { t.Fatalf("AcquireHandle status = %v, want OK", status) } diff --git a/weed/mount/weedfs_link.go b/weed/mount/weedfs_link.go index c2178c780..1797627ca 100644 --- a/weed/mount/weedfs_link.go +++ b/weed/mount/weedfs_link.go @@ -57,7 +57,9 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * // probe can move the two apart even for an entry that does carry one. sourceInode := in.Oldnodeid - oldEntry, _, status := wfs.maybeLoadEntry(oldEntryPath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + oldEntry, _, status := wfs.maybeLoadEntry(ctx, oldEntryPath) if status != fuse.OK { return status } @@ -89,7 +91,7 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * // Do not fall back to the pre-lock snapshot: if every alias // was deleted while we waited, abort instead of deriving the // next counter from a stale entry. - fresh, _, freshStatus := wfs.maybeLoadEntry(oldEntryPath) + fresh, _, freshStatus := wfs.maybeLoadEntry(ctx, oldEntryPath) if freshStatus != fuse.OK { return freshStatus } @@ -137,21 +139,20 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * // apply changes to the filer, and also apply to local metaCache wfs.mapPbIdFromLocalToFiler(request.Entry) - ctx := context.Background() - updateResp, err := wfs.streamUpdateEntry(ctx, updateOldEntryRequest) + updateResp, err := wfs.streamUpdateEntry(context.Background(), updateOldEntryRequest) if err == nil { updateEvent := updateResp.GetMetadataEvent() if updateEvent == nil { updateEvent = metadataUpdateEvent(oldParentPath, updateOldEntryRequest.Entry) } - if applyErr := wfs.applyLocalMetadataEvent(ctx, updateEvent); applyErr != nil { + if applyErr := wfs.applyLocalMetadataEvent(context.Background(), updateEvent); applyErr != nil { glog.Warningf("link %s: best-effort metadata apply failed: %v", oldEntryPath, applyErr) wfs.inodeToPath.InvalidateChildrenCache(util.FullPath(oldParentPath)) } } if err == nil { var createResp *filer_pb.CreateEntryResponse - createResp, err = wfs.streamCreateEntry(ctx, request) + createResp, err = wfs.streamCreateEntry(context.Background(), request) if err != nil { // Rollback: restore original HardLinkId/Counter on the source entry oldEntry.HardLinkId = origHardLinkId @@ -161,7 +162,7 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * Entry: oldEntry, Signatures: []int32{wfs.signature}, } - if _, rollbackErr := wfs.streamUpdateEntry(ctx, rollbackReq); rollbackErr != nil { + if _, rollbackErr := wfs.streamUpdateEntry(context.Background(), rollbackReq); rollbackErr != nil { glog.Warningf("link rollback %s: %v", oldEntryPath, rollbackErr) } } else { @@ -169,7 +170,7 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out * if createEvent == nil { createEvent = metadataCreateEvent(string(newParentPath), request.Entry) } - if applyErr := wfs.applyLocalMetadataEvent(ctx, createEvent); applyErr != nil { + if applyErr := wfs.applyLocalMetadataEvent(context.Background(), createEvent); applyErr != nil { glog.Warningf("link %s: best-effort metadata apply failed: %v", newParentPath.Child(name), applyErr) wfs.inodeToPath.InvalidateChildrenCache(newParentPath) } diff --git a/weed/mount/weedfs_link_test.go b/weed/mount/weedfs_link_test.go index d0872d182..e81bb5957 100644 --- a/weed/mount/weedfs_link_test.go +++ b/weed/mount/weedfs_link_test.go @@ -106,7 +106,7 @@ func TestLinkReportsKernelNodeIdWhenEntryHasNoPersistedInode(t *testing.T) { if _, status := wfs.inodeToPath.GetPath(sourceInode); status != fuse.OK { t.Fatalf("GetPath(%d) = %v, want OK", sourceInode, status) } - linked, _, status := wfs.maybeLoadEntry(sourcePath) + linked, _, status := wfs.maybeLoadEntry(context.Background(), sourcePath) if status != fuse.OK { t.Fatalf("reload source: %v", status) } @@ -229,7 +229,7 @@ func TestLinkSyncsHardLinkSiblingsOnThirdLink(t *testing.T) { // any sibling returns the attributes of the last write to any of them, // synced or not. Feeding the loop an authoritative counter that no other // write has stored is what makes its work visible. - authoritative, _, status := wfs.maybeLoadEntry(sourcePath) + authoritative, _, status := wfs.maybeLoadEntry(context.Background(), sourcePath) if status != fuse.OK { t.Fatalf("reload source: %v", status) } diff --git a/weed/mount/weedfs_posix_lock_routed.go b/weed/mount/weedfs_posix_lock_routed.go index 89d311ca6..ae7a48ce8 100644 --- a/weed/mount/weedfs_posix_lock_routed.go +++ b/weed/mount/weedfs_posix_lock_routed.go @@ -94,7 +94,7 @@ func (wfs *WFS) posixLockKeyForInode(inode uint64) (string, bool) { if status != fuse.OK { return "", false } - if entry, _, st := wfs.maybeLoadEntry(path); st == fuse.OK && entry != nil && len(entry.HardLinkId) > 0 { + if entry, _, st := wfs.maybeLoadEntry(context.Background(), path); st == fuse.OK && entry != nil && len(entry.HardLinkId) > 0 { return posixLockKeyPrefix + "hl:" + hex.EncodeToString(entry.HardLinkId), true } return posixLockKeyPrefix + string(path), true diff --git a/weed/mount/weedfs_rename.go b/weed/mount/weedfs_rename.go index cd0ae361b..369314edf 100644 --- a/weed/mount/weedfs_rename.go +++ b/weed/mount/weedfs_rename.go @@ -231,13 +231,17 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string } newPath := newDir.Child(newName) - oldEntry, _, status := wfs.maybeLoadEntry(oldPath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + oldEntry, _, status := wfs.maybeLoadEntry(ctx, oldPath) if status != fuse.OK { return status } // POSIX: enforce sticky bit on the source directory. - if oldDirEntry, _, dirCode := wfs.maybeLoadEntry(oldDir); dirCode == fuse.OK && oldDirEntry != nil && oldDirEntry.Attributes != nil { + if oldDirEntry, _, dirCode := wfs.maybeLoadEntry(ctx, oldDir); dirCode != fuse.OK { + return dirCode + } else if oldDirEntry != nil && oldDirEntry.Attributes != nil { targetUid := uint32(0) if oldEntry != nil && oldEntry.Attributes != nil { targetUid = oldEntry.Attributes.Uid @@ -249,14 +253,18 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string var newEntry *filer_pb.Entry if in.Flags != RenameNoReplace { - if loaded, _, newStatus := wfs.maybeLoadEntry(newPath); newStatus == fuse.OK { + if loaded, _, newStatus := wfs.maybeLoadEntry(ctx, newPath); newStatus == fuse.OK { newEntry = loaded + } else if newStatus != fuse.ENOENT { + return newStatus } } // POSIX: enforce sticky bit on the destination directory when replacing an existing entry. if newEntry != nil { - if newDirEntry, _, dirCode := wfs.maybeLoadEntry(newDir); dirCode == fuse.OK && newDirEntry != nil && newDirEntry.Attributes != nil { + if newDirEntry, _, dirCode := wfs.maybeLoadEntry(ctx, newDir); dirCode != fuse.OK { + return dirCode + } else if newDirEntry != nil && newDirEntry.Attributes != nil { targetUid := uint32(0) if newEntry.Attributes != nil { targetUid = newEntry.Attributes.Uid @@ -391,13 +399,13 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string // skips old-path metadata creation (which would re-insert the // renamed entry into the meta cache after rename events clean it up). wfs.fhMap.MarkInodeRenamed(sourceInode) - wfs.waitForPendingAsyncFlush(sourceInode) + wfs.waitForPendingAsyncFlush(context.Background(), sourceInode) } else if oldEntry != nil && oldEntry.Attributes != nil && oldEntry.Attributes.Inode != 0 { // GetInode failed (Forget already removed the mapping), but the // entry's stored inode can still identify pending async flushes. sourceInode = oldEntry.Attributes.Inode wfs.fhMap.MarkInodeRenamed(sourceInode) - wfs.waitForPendingAsyncFlush(sourceInode) + wfs.waitForPendingAsyncFlush(context.Background(), sourceInode) } // Replacing the destination deletes what it held, so a handle still open @@ -417,7 +425,7 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string // belongs to the destination as it stands, and one that runs // marked is skipped and takes its handle with it, past the reach // of the restore below. - wfs.waitForPendingAsyncFlush(targetInode) + wfs.waitForPendingAsyncFlush(context.Background(), targetInode) epoch := wfs.markHandleDeleted(targetInode) // A rename that does not happen leaves the destination where it // was, so its handle has to go back to writing that name. @@ -438,8 +446,7 @@ func (wfs *WFS) Rename(cancel <-chan struct{}, in *fuse.RenameIn, oldName string Signatures: []int32{wfs.signature}, } - ctx := context.Background() - err := wfs.doRename(ctx, request, oldPath, newPath, &newPathLock) + err := wfs.doRename(context.Background(), request, oldPath, newPath, &newPathLock) if err != nil { glog.V(0).Infof("Rename %s => %s: %v", oldPath, newPath, err) // Map error strings to FUSE status codes. String matching is used diff --git a/weed/mount/weedfs_symlink.go b/weed/mount/weedfs_symlink.go index d9a413d4f..bea946a3b 100644 --- a/weed/mount/weedfs_symlink.go +++ b/weed/mount/weedfs_symlink.go @@ -91,7 +91,9 @@ func (wfs *WFS) Readlink(cancel <-chan struct{}, header *fuse.InHeader) (out []b return } - entry, _, status := wfs.maybeLoadEntry(entryFullPath) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + entry, _, status := wfs.maybeLoadEntry(ctx, entryFullPath) if status != fuse.OK { return nil, status } diff --git a/weed/mount/weedfs_xattr.go b/weed/mount/weedfs_xattr.go index 07ee81032..da3815ccb 100644 --- a/weed/mount/weedfs_xattr.go +++ b/weed/mount/weedfs_xattr.go @@ -38,7 +38,9 @@ func (wfs *WFS) GetXAttr(cancel <-chan struct{}, header *fuse.InHeader, attr str return 0, fuse.EINVAL } - _, _, entry, status := wfs.maybeReadEntry(header.NodeId) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + _, _, entry, status := wfs.maybeReadEntry(ctx, header.NodeId) if status != fuse.OK { return 0, status } @@ -105,7 +107,9 @@ func (wfs *WFS) SetXAttr(cancel <-chan struct{}, input *fuse.SetXAttrIn, attr st } } - path, fh, entry, status := wfs.maybeReadEntry(input.NodeId) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + path, fh, entry, status := wfs.maybeReadEntry(ctx, input.NodeId) if status != fuse.OK { return status } @@ -160,7 +164,9 @@ func (wfs *WFS) ListXAttr(cancel <-chan struct{}, header *fuse.InHeader, dest [] return 0, fuse.Status(syscall.ENOTSUP) } - _, _, entry, status := wfs.maybeReadEntry(header.NodeId) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + _, _, entry, status := wfs.maybeReadEntry(ctx, header.NodeId) if status != fuse.OK { return 0, status } @@ -197,7 +203,9 @@ func (wfs *WFS) RemoveXAttr(cancel <-chan struct{}, header *fuse.InHeader, attr if len(attr) == 0 { return fuse.EINVAL } - path, fh, entry, status := wfs.maybeReadEntry(header.NodeId) + ctx, cancelFunc := ctxFromCancel(cancel) + defer cancelFunc() + path, fh, entry, status := wfs.maybeReadEntry(ctx, header.NodeId) if status != fuse.OK { return status } diff --git a/weed/mount/weedfs_xattr_cancel_test.go b/weed/mount/weedfs_xattr_cancel_test.go new file mode 100644 index 000000000..1a7c1b835 --- /dev/null +++ b/weed/mount/weedfs_xattr_cancel_test.go @@ -0,0 +1,96 @@ +package mount + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// stalledLookupFiler parks every LookupDirectoryEntry until the caller's +// context ends, the way a wedged filer holds an RPC forever. +type stalledLookupFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + lookupStarted chan struct{} +} + +func (s *stalledLookupFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + select { + case <-s.lookupStarted: + default: + close(s.lookupStarted) + } + <-ctx.Done() + return nil, status.Error(codes.Canceled, ctx.Err().Error()) +} + +// A FUSE interrupt must unwind a metadata lookup stuck on the filer; without +// it the op waits on the RPC forever and the caller — Samba issuing +// listxattr — never sees an answer. +func TestListXAttrCanceledWhileFilerStalled(t *testing.T) { + wfs := newInvalidateTestWFS(t) + fake := &stalledLookupFiler{lookupStarted: make(chan struct{})} + startFakeFiler(t, wfs, fake) + + wfs.inodeToPath.EnsurePath(util.FullPath("/dir"), true) + filePath := util.FullPath("/dir/f") + inode := wfs.inodeToPath.AllocateInode(filePath, time.Now().Unix()) + wfs.inodeToPath.AddPath(inode, filePath) + + cancel := make(chan struct{}) + done := make(chan fuse.Status, 1) + go func() { + _, code := wfs.ListXAttr(cancel, &fuse.InHeader{NodeId: inode}, nil) + done <- code + }() + + select { + case <-fake.lookupStarted: + case <-time.After(10 * time.Second): + t.Fatal("lookup never reached the filer") + } + close(cancel) + + select { + case code := <-done: + if code != fuse.EINTR { + t.Fatalf("interrupted lookup returned %v, want EINTR", code) + } + case <-time.After(10 * time.Second): + t.Fatal("ListXAttr still blocked after the FUSE request was interrupted") + } +} + +// Even without an interrupt the filer lookup has a deadline, so a hung RPC +// returns a filesystem error instead of parking the caller indefinitely. +func TestListXAttrTimesOutWhileFilerStalled(t *testing.T) { + wfs := newInvalidateTestWFS(t) + wfs.filerCallTimeout = 50 * time.Millisecond + fake := &stalledLookupFiler{lookupStarted: make(chan struct{})} + startFakeFiler(t, wfs, fake) + + wfs.inodeToPath.EnsurePath(util.FullPath("/dir"), true) + filePath := util.FullPath("/dir/f") + inode := wfs.inodeToPath.AllocateInode(filePath, time.Now().Unix()) + wfs.inodeToPath.AddPath(inode, filePath) + + done := make(chan fuse.Status, 1) + go func() { + _, code := wfs.ListXAttr(nil, &fuse.InHeader{NodeId: inode}, nil) + done <- code + }() + + select { + case code := <-done: + if code != fuse.EIO { + t.Fatalf("stalled filer lookup returned %v, want EIO", code) + } + case <-time.After(30 * time.Second): + t.Fatal("ListXAttr still blocked after the filer lookup deadline") + } +}