Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-13 01:28:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

mount: unwind metadata lookups on FUSE interrupt (#11717)

Browse Source 
* 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>
This commit is contained in:
Chris LuandDevin authored and GitHub committed 2026-10-12 09:42:52 +08:00
1 parent 50a9312e1e
commit bd0bc06df7
26 files changed
+337 -139
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

weed/mount/dir_sink.go
+3 -1
View File
@@ -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)
}
weed/mount/meta_cache/meta_cache_build_test.go
+5 -5
View File
@@ -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")) {
weed/mount/meta_cache/meta_cache_init.go
+12 -5
View File
@@ -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 {
weed/mount/meta_cache/oversized_dir_test.go
+5 -5
View File
@@ -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")) {
weed/mount/weedfs.go
+37 -12
View File
@@ -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",
weed/mount/weedfs_access.go
+3 -1
View File
@@ -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
}
weed/mount/weedfs_attr.go
+7 -3
View File
@@ -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
}
weed/mount/weedfs_dir_lookup.go
+3 -1
View File
@@ -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
}
weed/mount/weedfs_dir_lookup_cache_test.go
+8 -8
View File
@@ -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 {
weed/mount/weedfs_dir_mkrm.go
+9 -2
View File
@@ -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
weed/mount/weedfs_dir_read.go
+44 -19
View File
@@ -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
}
weed/mount/weedfs_dir_read_bench_test.go
+1 -1
View File
@@ -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,
weed/mount/weedfs_dir_read_pagination_test.go
+4 -4
View File
@@ -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,
weed/mount/weedfs_file_io.go
+3 -1
View File
@@ -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
weed/mount/weedfs_file_mkrm.go
+22 -14
View File
@@ -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
weed/mount/weedfs_file_mkrm_test.go
+2 -2
View File
@@ -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")
}
}
weed/mount/weedfs_file_read.go
+2 -11
View File
@@ -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)
weed/mount/weedfs_filehandle.go
+16 -7
View File
@@ -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
}
}
weed/mount/weedfs_invalidate_open_handle_test.go
+12 -12
View File
@@ -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)
}
weed/mount/weedfs_link.go
+9 -8
View File
@@ -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)
}
weed/mount/weedfs_link_test.go
+2 -2
View File
@@ -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)
}
weed/mount/weedfs_posix_lock_routed.go
+1 -1
View File
@@ -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
weed/mount/weedfs_rename.go
+16 -9
View File
@@ -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
weed/mount/weedfs_symlink.go
+3 -1
View File
@@ -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
}
weed/mount/weedfs_xattr.go
+12 -4
View File
@@ -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
}
weed/mount/weedfs_xattr_cancel_test.go
+96
View File
@@ -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")
}
}