mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
filer: skip the lazy remote fetch only in CreateEntry's lookup inside a transaction
The guard on every lookup inside a store transaction hid a source or target the filer holds only on the remote from the rename, and the prefetch before the transaction did not reach the children of a directory rename. The deadlock needs a write: CreateEntry's lookup after the transaction deleted the path. That lookup alone skips the fetch; reads inside the transaction keep it, and their persist on a second connection holds no lock the transaction waits for. (cherry picked from commit f555092c324780a8fe5314b238694551e3ef6be5)
This commit is contained in:
1 parent
26d1dce6f7
commit
b7dfff6616
4 files changed
+24
-65
No files matched your search
+10
-1
@@ -516,7 +516,16 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry,
|
||||
oldEntry := existing
|
||||
if oldEntry == nil {
|
||||
var findErr error
|
||||
oldEntry, findErr = f.FindEntry(ctx, entry.FullPath)
|
||||
// Inside a store transaction the lookup must not run the lazy remote
|
||||
// fetch: the fetch persists on a second connection, which waits on
|
||||
// the row lock this transaction holds when it deleted the path (a
|
||||
// rename onto an existing target). The create writes the path either
|
||||
// way; reads inside the transaction keep the fetch.
|
||||
findCtx := ctx
|
||||
if ctx.Value("tx") != nil {
|
||||
findCtx = context.WithValue(ctx, lazyFetchContextKey{}, true)
|
||||
}
|
||||
oldEntry, findErr = f.FindEntry(findCtx, entry.FullPath)
|
||||
if findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound) {
|
||||
// A create cannot decide whether the path exists when the lookup
|
||||
// itself failed; proceeding would upsert over it.
|
||||
|
||||
@@ -41,12 +41,6 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) (
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// A lookup inside a store transaction persists on a second connection,
|
||||
// which waits on the row locks the transaction holds.
|
||||
if ctx.Value("tx") != nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if f.RemoteStorage == nil {
|
||||
return nil, nil
|
||||
}
|
||||
@@ -438,17 +432,3 @@ func (f *Filer) RemoteCopyLocations(ctx context.Context, entry *Entry, oldPath,
|
||||
dst = MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath)
|
||||
return src, dst, copier, nil
|
||||
}
|
||||
|
||||
// PrefetchRemoteEntries runs the lazy remote fetch for the paths a store
|
||||
// transaction is about to read. Inside the transaction the fetch is off,
|
||||
// so an object the filer has not seen yet would read as absent there.
|
||||
func (f *Filer) PrefetchRemoteEntries(ctx context.Context, paths ...util.FullPath) {
|
||||
if ctx.Value("tx") != nil {
|
||||
return
|
||||
}
|
||||
for _, p := range paths {
|
||||
if _, err := f.FindEntry(ctx, p); err != nil && !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
glog.V(1).InfofCtx(ctx, "prefetch remote entry %s: %v", p, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -734,59 +734,34 @@ func TestMaybeLazyFetchFromRemote_ContextGuardPreventsRecursion(t *testing.T) {
|
||||
assert.Equal(t, 0, countingStub.statCalls, "guard should prevent StatFile from being called")
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_SkipsInsideStoreTransaction(t *testing.T) {
|
||||
func TestCreateEntry_SkipsTheLazyFetchInsideAStoreTransaction(t *testing.T) {
|
||||
const storageType = "stub_lazy_tx"
|
||||
countingStub := &countingRemoteClient{
|
||||
stubRemoteClient: stubRemoteClient{
|
||||
statResult: &filer_pb.RemoteEntry{RemoteMtime: 1, RemoteSize: 1},
|
||||
},
|
||||
}
|
||||
f, _ := newMountedTestFiler(t, storageType, countingStub, 0)
|
||||
|
||||
// the key a store transaction travels under (abstract_sql getTxOrDB)
|
||||
txCtx := context.WithValue(context.Background(), "tx", struct{}{})
|
||||
entry, err := f.maybeLazyFetchFromRemote(txCtx, "/buckets/mybucket/dir/a.txt")
|
||||
require.NoError(t, err)
|
||||
assert.Nil(t, entry)
|
||||
assert.Equal(t, 0, countingStub.statCalls, "a lookup inside a transaction must not consult the remote")
|
||||
|
||||
entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/a.txt")
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, entry)
|
||||
assert.NotZero(t, countingStub.statCalls, "the same lookup outside a transaction consults the remote")
|
||||
}
|
||||
|
||||
func TestPrefetchRemoteEntries_MaterializesBeforeATransaction(t *testing.T) {
|
||||
const storageType = "stub_lazy_prefetch"
|
||||
countingStub := &countingRemoteClient{
|
||||
stubRemoteClient: stubRemoteClient{
|
||||
statResult: &filer_pb.RemoteEntry{RemoteMtime: 1, RemoteSize: 1},
|
||||
},
|
||||
}
|
||||
f, store := newMountedTestFiler(t, storageType, countingStub, 0)
|
||||
putDirEntry(store, "/buckets/mybucket/dir")
|
||||
target := util.FullPath("/buckets/mybucket/dir/target.txt")
|
||||
// the key a store transaction travels under (abstract_sql getTxOrDB)
|
||||
txCtx := context.WithValue(context.Background(), "tx", struct{}{})
|
||||
|
||||
// unseen on the filer, present on the remote: inside a transaction it
|
||||
// reads as absent
|
||||
_, err := f.FindEntry(txCtx, target)
|
||||
require.ErrorIs(t, err, filer_pb.ErrNotFound)
|
||||
assert.Equal(t, 0, countingStub.statCalls)
|
||||
// a create inside the transaction: its lookup does not consult the
|
||||
// remote, the entry is inserted as given
|
||||
created := &Entry{FullPath: "/buckets/mybucket/dir/a.txt", Attr: Attr{Mode: 0644, Mtime: time.Now(), Crtime: time.Now()}}
|
||||
require.NoError(t, f.CreateEntry(txCtx, created, nil, false, false, nil, false, f.MaxFilenameLength))
|
||||
assert.Equal(t, 0, countingStub.statCalls, "a create inside a transaction must not consult the remote")
|
||||
got, err := f.FindEntry(txCtx, "/buckets/mybucket/dir/a.txt")
|
||||
require.NoError(t, err)
|
||||
assert.False(t, got.IsInRemoteOnly())
|
||||
|
||||
// the prefetch outside the transaction materializes it, and the
|
||||
// transaction then finds it in the store
|
||||
f.PrefetchRemoteEntries(context.Background(), target, "/buckets/mybucket/dir/absent.txt")
|
||||
assert.NotZero(t, countingStub.statCalls)
|
||||
entry, err := f.FindEntry(txCtx, target)
|
||||
// a read inside the transaction keeps the fetch: a target the filer only
|
||||
// holds on the remote is still seen by a rename's target check
|
||||
entry, err := f.FindEntry(txCtx, "/buckets/mybucket/dir/b.txt")
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, entry)
|
||||
assert.True(t, entry.IsInRemoteOnly())
|
||||
|
||||
// a prefetch that itself runs inside a transaction does nothing
|
||||
before := countingStub.statCalls
|
||||
f.PrefetchRemoteEntries(txCtx, "/buckets/mybucket/dir/other.txt")
|
||||
assert.Equal(t, before, countingStub.statCalls)
|
||||
assert.NotZero(t, countingStub.statCalls)
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_GuardedClientRejectsEndpoint(t *testing.T) {
|
||||
|
||||
@@ -47,10 +47,6 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
|
||||
|
||||
defer fs.acquireRenamePathLocks("AtomicRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))()
|
||||
|
||||
// the lazy remote fetch is off inside the transaction: a source or
|
||||
// target the filer only holds on the remote is materialized first
|
||||
fs.filer.PrefetchRemoteEntries(ctx, oldParent.Child(req.OldName), newParent.Child(req.NewName))
|
||||
|
||||
// a SQL store keeps the transaction in the context and refuses calls
|
||||
// through it once committed; the work after the commit uses this one
|
||||
afterCommitCtx := ctx
|
||||
@@ -122,7 +118,6 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
|
||||
defer fs.acquireRenamePathLocks("StreamRenameEntry", oldParent.Child(req.OldName), newParent.Child(req.NewName))()
|
||||
|
||||
ctx := context.Background()
|
||||
fs.filer.PrefetchRemoteEntries(ctx, oldParent.Child(req.OldName), newParent.Child(req.NewName))
|
||||
|
||||
afterCommitCtx := ctx
|
||||
ctx, err = fs.filer.BeginTransaction(ctx)
|
||||
|
||||
Reference in new issue
Block a user