From bd0bc06df77f06f74a6b04a6d8955f61f8063989 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 12 Oct 2026 09:42:52 +0800 Subject: [PATCH] mount: unwind metadata lookups on FUSE interrupt (#11717) * mount: propagate request context through metadata lookups The FUSE interrupt channel never reached the filer RPC: lookupEntry ran its LookupEntry under context.Background(), so a stalled filer parked ops like ListXAttr forever even after the kernel gave up on them. Thread a context from each op's cancel channel through maybeReadEntry, maybeLoadEntry, lookupEntry, ensureDirectoryVisited and waitForPendingAsyncFlush; the existing withFilerClient attempt bound turns it into both an interrupt and a deadline. A caller cancellation now surfaces as EINTR instead of waiting out the RPC. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: reuse ctxFromCancel for the read path's interrupt bridge Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: regression tests for canceled and stalled filer lookups Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: thread ctx through AcquireHandle Open, Create and Mknod all carry the request's cancel channel; let it bound the pending-flush wait and metadata load inside AcquireHandle too. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: keep flush waits and write sequences uninterruptible The cancel-derived context must only cover metadata reads: an interrupted Unlink or Rename that skips the pending-flush wait can race the async write back into existence, and Link/Rename mutations that run under a canceled context abort mid-sequence, strand the rollback, and leave wrong hard-link counts. Restore the blocking flush wait and run each op's write phase on context.Background(). Rmdir and Unlink also propagate EINTR from the parent lookup instead of proceeding past a permission check the interrupt prevented. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: interrupt shared directory builds and directory reads Waiters behind another request's singleflight directory build blocked until it finished even when interrupted; switch the wait to DoChan so the request context can abandon it. ReadDir paths now carry the request context through EnsureListingFresh, cache listings and direct reads, and map its cancellation to EINTR rather than EIO. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: assert exact statuses in filer-stall regression tests Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: let read-side flush waits release on interrupt waitForPendingAsyncFlush reports whether the flush actually finished: lookupEntry and AcquireHandle return EINTR when the request is abandoned mid-wait, while Unlink and Rename keep waiting on context.Background() so the delete/rename still sees the flushed state. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: propagate interrupt through remaining lookup checks Rename's source and destination permission lookups now return EINTR on interrupt instead of skipping checkStickyBit or masking an existing target, and the readdir preload visit maps cancellation to EINTR like the other branches. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * mount: fail closed on errored rename prerequisite lookups * mount: stop rmdir on interrupted target lookup, fail closed on dir lookup errors --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/mount/dir_sink.go | 4 +- .../mount/meta_cache/meta_cache_build_test.go | 10 +- weed/mount/meta_cache/meta_cache_init.go | 17 +++- weed/mount/meta_cache/oversized_dir_test.go | 10 +- weed/mount/weedfs.go | 49 +++++++--- weed/mount/weedfs_access.go | 4 +- weed/mount/weedfs_attr.go | 10 +- weed/mount/weedfs_dir_lookup.go | 4 +- weed/mount/weedfs_dir_lookup_cache_test.go | 16 ++-- weed/mount/weedfs_dir_mkrm.go | 11 ++- weed/mount/weedfs_dir_read.go | 63 ++++++++---- weed/mount/weedfs_dir_read_bench_test.go | 2 +- weed/mount/weedfs_dir_read_pagination_test.go | 8 +- weed/mount/weedfs_file_io.go | 4 +- weed/mount/weedfs_file_mkrm.go | 36 ++++--- weed/mount/weedfs_file_mkrm_test.go | 4 +- weed/mount/weedfs_file_read.go | 13 +-- weed/mount/weedfs_filehandle.go | 23 +++-- .../weedfs_invalidate_open_handle_test.go | 24 ++--- weed/mount/weedfs_link.go | 17 ++-- weed/mount/weedfs_link_test.go | 4 +- weed/mount/weedfs_posix_lock_routed.go | 2 +- weed/mount/weedfs_rename.go | 25 +++-- weed/mount/weedfs_symlink.go | 4 +- weed/mount/weedfs_xattr.go | 16 +++- weed/mount/weedfs_xattr_cancel_test.go | 96 +++++++++++++++++++ 26 files changed, 337 insertions(+), 139 deletions(-) create mode 100644 weed/mount/weedfs_xattr_cancel_test.go 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") + } +}