mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
filer: materialize remote-only rename paths before the transaction
With the lazy fetch off inside a store transaction, a source or target the filer holds only on the remote would read as absent to AtomicRenameEntry and StreamRenameEntry: a rename with FailIfTargetExists could overwrite it, and a remote-only source would skip the deferred copy. PrefetchRemoteEntries runs the fetch for both paths before BeginTransaction, under the path locks. (cherry picked from commit e61f9718018669ef59768cd5eda25b6466d69597)
This commit is contained in:
1 parent
5dd033fc3e
commit
26d1dce6f7
3 files changed
+52
No files matched your search
@@ -438,3 +438,17 @@ 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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -756,6 +756,39 @@ func TestMaybeLazyFetchFromRemote_SkipsInsideStoreTransaction(t *testing.T) {
|
||||
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")
|
||||
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)
|
||||
|
||||
// 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)
|
||||
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)
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_GuardedClientRejectsEndpoint(t *testing.T) {
|
||||
const storageType = "stub_lazy_guarded"
|
||||
countingStub := &countingRemoteClient{
|
||||
|
||||
@@ -47,6 +47,10 @@ 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
|
||||
@@ -118,6 +122,7 @@ 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