Use filer-side copy for mounted whole-file copy_file_range (#8747)
* Optimize mounted whole-file copy_file_range * Address mounted copy review feedback * Harden mounted copy fast path --------- Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
Chris Lu
@@ -19,6 +19,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/mount_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/security"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
|
||||
@@ -30,27 +31,29 @@ import (
|
||||
)
|
||||
|
||||
type Option struct {
|
||||
filerIndex int32 // align memory for atomic read/write
|
||||
FilerAddresses []pb.ServerAddress
|
||||
MountDirectory string
|
||||
GrpcDialOption grpc.DialOption
|
||||
FilerMountRootPath string
|
||||
Collection string
|
||||
Replication string
|
||||
TtlSec int32
|
||||
DiskType types.DiskType
|
||||
ChunkSizeLimit int64
|
||||
ConcurrentWriters int
|
||||
ConcurrentReaders int
|
||||
CacheDirForRead string
|
||||
CacheSizeMBForRead int64
|
||||
CacheDirForWrite string
|
||||
CacheMetaTTlSec int
|
||||
DataCenter string
|
||||
Umask os.FileMode
|
||||
Quota int64
|
||||
DisableXAttr bool
|
||||
IsMacOs bool
|
||||
filerIndex int32 // align memory for atomic read/write
|
||||
FilerAddresses []pb.ServerAddress
|
||||
MountDirectory string
|
||||
GrpcDialOption grpc.DialOption
|
||||
FilerSigningKey security.SigningKey
|
||||
FilerSigningExpiresAfterSec int
|
||||
FilerMountRootPath string
|
||||
Collection string
|
||||
Replication string
|
||||
TtlSec int32
|
||||
DiskType types.DiskType
|
||||
ChunkSizeLimit int64
|
||||
ConcurrentWriters int
|
||||
ConcurrentReaders int
|
||||
CacheDirForRead string
|
||||
CacheSizeMBForRead int64
|
||||
CacheDirForWrite string
|
||||
CacheMetaTTlSec int
|
||||
DataCenter string
|
||||
Umask os.FileMode
|
||||
Quota int64
|
||||
DisableXAttr bool
|
||||
IsMacOs bool
|
||||
|
||||
MountUid uint32
|
||||
MountGid uint32
|
||||
@@ -173,11 +176,11 @@ func NewSeaweedFileSystem(option *Option) *WFS {
|
||||
dhMap: NewDirectoryHandleToInode(),
|
||||
filerClient: filerClient, // nil for proxy mode, initialized for direct access
|
||||
pendingAsyncFlush: make(map[uint64]chan struct{}),
|
||||
fhLockTable: util.NewLockTable[FileHandleId](),
|
||||
refreshingDirs: make(map[util.FullPath]struct{}),
|
||||
dirHotWindow: dirHotWindow,
|
||||
dirHotThreshold: dirHotThreshold,
|
||||
dirIdleEvict: dirIdleEvict,
|
||||
fhLockTable: util.NewLockTable[FileHandleId](),
|
||||
refreshingDirs: make(map[util.FullPath]struct{}),
|
||||
dirHotWindow: dirHotWindow,
|
||||
dirHotThreshold: dirHotThreshold,
|
||||
dirIdleEvict: dirIdleEvict,
|
||||
}
|
||||
|
||||
wfs.option.filerIndex = int32(rand.IntN(len(option.FilerAddresses)))
|
||||
|
||||
@@ -1,15 +1,63 @@
|
||||
package mount
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/go-fuse/v2/fuse"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/security"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
|
||||
request_id "github.com/seaweedfs/seaweedfs/weed/util/request_id"
|
||||
)
|
||||
|
||||
type serverSideWholeFileCopyOutcome uint8
|
||||
|
||||
const (
|
||||
serverSideWholeFileCopyNotCommitted serverSideWholeFileCopyOutcome = iota
|
||||
serverSideWholeFileCopyCommitted
|
||||
serverSideWholeFileCopyAmbiguous
|
||||
)
|
||||
|
||||
type wholeFileServerCopyRequest struct {
|
||||
srcPath util.FullPath
|
||||
dstPath util.FullPath
|
||||
sourceSize int64
|
||||
srcInode uint64
|
||||
srcMtime int64
|
||||
dstInode uint64
|
||||
dstMtime int64
|
||||
dstSize int64
|
||||
sourceMime string
|
||||
sourceMd5 []byte
|
||||
copyRequestID string
|
||||
}
|
||||
|
||||
// performServerSideWholeFileCopy is a package-level seam so tests can override
|
||||
// the filer call without standing up an HTTP endpoint.
|
||||
var performServerSideWholeFileCopy = func(cancel <-chan struct{}, wfs *WFS, copyRequest wholeFileServerCopyRequest) (*filer_pb.Entry, serverSideWholeFileCopyOutcome, error) {
|
||||
return wfs.copyEntryViaFiler(cancel, copyRequest)
|
||||
}
|
||||
|
||||
// filerCopyRequestTimeout bounds the mount->filer POST so a stalled copy does
|
||||
// not block copy_file_range workers indefinitely.
|
||||
const filerCopyRequestTimeout = 60 * time.Second
|
||||
|
||||
// filerCopyReadbackTimeout gives the follow-up metadata reload a fresh deadline
|
||||
// after the filer already accepted the copy request.
|
||||
const filerCopyReadbackTimeout = 15 * time.Second
|
||||
|
||||
// CopyFileRange copies data from one file to another from and to specified offsets.
|
||||
//
|
||||
// See https://man7.org/linux/man-pages/man2/copy_file_range.2.html
|
||||
@@ -70,6 +118,10 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn)
|
||||
in.OffOut, in.OffOut+in.Len,
|
||||
)
|
||||
|
||||
if written, handled, status := wfs.tryServerSideWholeFileCopy(cancel, in, fhIn, fhOut); handled {
|
||||
return written, status
|
||||
}
|
||||
|
||||
// Concurrent copy operations could allocate too much memory, so we want to
|
||||
// throttle our concurrency, scaling with the number of writers the mount
|
||||
// was configured with.
|
||||
@@ -155,3 +207,306 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn)
|
||||
written = uint32(totalCopied)
|
||||
return written, fuse.OK
|
||||
}
|
||||
|
||||
func (wfs *WFS) tryServerSideWholeFileCopy(cancel <-chan struct{}, in *fuse.CopyFileRangeIn, fhIn, fhOut *FileHandle) (written uint32, handled bool, code fuse.Status) {
|
||||
copyRequest, ok := wholeFileServerCopyCandidate(fhIn, fhOut, in)
|
||||
if !ok {
|
||||
return 0, false, fuse.OK
|
||||
}
|
||||
|
||||
glog.V(1).Infof("CopyFileRange server-side copy %s => %s (%d bytes)", copyRequest.srcPath, copyRequest.dstPath, copyRequest.sourceSize)
|
||||
|
||||
entry, outcome, err := performServerSideWholeFileCopy(cancel, wfs, copyRequest)
|
||||
switch outcome {
|
||||
case serverSideWholeFileCopyCommitted:
|
||||
if err != nil {
|
||||
glog.Warningf("CopyFileRange server-side copy %s => %s committed but local refresh failed: %v", copyRequest.srcPath, copyRequest.dstPath, err)
|
||||
} else {
|
||||
glog.V(1).Infof("CopyFileRange server-side copy %s => %s completed (%d bytes)", copyRequest.srcPath, copyRequest.dstPath, copyRequest.sourceSize)
|
||||
}
|
||||
wfs.applyServerSideWholeFileCopyResult(fhIn, fhOut, copyRequest.dstPath, entry, copyRequest.sourceSize)
|
||||
return uint32(copyRequest.sourceSize), true, fuse.OK
|
||||
case serverSideWholeFileCopyAmbiguous:
|
||||
glog.Warningf("CopyFileRange server-side copy %s => %s outcome ambiguous: %v", copyRequest.srcPath, copyRequest.dstPath, err)
|
||||
return 0, true, fuse.EIO
|
||||
default:
|
||||
glog.V(0).Infof("CopyFileRange server-side copy %s => %s fallback to chunk copy: %v", copyRequest.srcPath, copyRequest.dstPath, err)
|
||||
return 0, false, fuse.OK
|
||||
}
|
||||
}
|
||||
|
||||
func (wfs *WFS) applyServerSideWholeFileCopyResult(fhIn, fhOut *FileHandle, dstPath util.FullPath, entry *filer_pb.Entry, sourceSize int64) {
|
||||
if entry == nil {
|
||||
entry = synthesizeLocalEntryForServerSideWholeFileCopy(fhIn, fhOut, sourceSize)
|
||||
}
|
||||
if entry == nil {
|
||||
glog.Warningf("CopyFileRange server-side copy %s left no local entry to apply", dstPath)
|
||||
return
|
||||
}
|
||||
|
||||
fhOut.SetEntry(entry)
|
||||
fhOut.RememberPath(dstPath)
|
||||
if entry.Attributes != nil {
|
||||
fhOut.contentType = entry.Attributes.Mime
|
||||
}
|
||||
fhOut.dirtyMetadata = false
|
||||
wfs.updateServerSideWholeFileCopyMetaCache(dstPath, entry)
|
||||
wfs.invalidateCopyDestinationCache(fhOut.inode, dstPath)
|
||||
}
|
||||
|
||||
func (wfs *WFS) updateServerSideWholeFileCopyMetaCache(dstPath util.FullPath, entry *filer_pb.Entry) {
|
||||
if wfs.metaCache == nil || entry == nil {
|
||||
return
|
||||
}
|
||||
|
||||
dir, _ := dstPath.DirAndName()
|
||||
event := metadataUpdateEvent(dir, entry)
|
||||
if applyErr := wfs.applyLocalMetadataEvent(context.Background(), event); applyErr != nil {
|
||||
glog.Warningf("CopyFileRange metadata update %s: %v", dstPath, applyErr)
|
||||
wfs.markDirectoryReadThrough(util.FullPath(dir))
|
||||
}
|
||||
}
|
||||
|
||||
func synthesizeLocalEntryForServerSideWholeFileCopy(fhIn, fhOut *FileHandle, sourceSize int64) *filer_pb.Entry {
|
||||
dstEntry := fhOut.GetEntry().GetEntry()
|
||||
if dstEntry == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
localEntry := proto.Clone(dstEntry).(*filer_pb.Entry)
|
||||
if localEntry.Attributes == nil {
|
||||
localEntry.Attributes = &filer_pb.FuseAttributes{}
|
||||
}
|
||||
|
||||
if srcEntry := fhIn.GetEntry().GetEntry(); srcEntry != nil {
|
||||
srcEntryCopy := proto.Clone(srcEntry).(*filer_pb.Entry)
|
||||
localEntry.Content = srcEntryCopy.Content
|
||||
localEntry.Chunks = srcEntryCopy.Chunks
|
||||
if srcEntryCopy.Attributes != nil {
|
||||
localEntry.Attributes.Mime = srcEntryCopy.Attributes.Mime
|
||||
localEntry.Attributes.Md5 = srcEntryCopy.Attributes.Md5
|
||||
}
|
||||
}
|
||||
|
||||
localEntry.Attributes.FileSize = uint64(sourceSize)
|
||||
localEntry.Attributes.Mtime = time.Now().Unix()
|
||||
return localEntry
|
||||
}
|
||||
|
||||
func wholeFileServerCopyCandidate(fhIn, fhOut *FileHandle, in *fuse.CopyFileRangeIn) (copyRequest wholeFileServerCopyRequest, ok bool) {
|
||||
if fhIn == nil || fhOut == nil || in == nil {
|
||||
glog.V(4).Infof("server-side copy: skipped (nil handle or input)")
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
if fhIn.fh == fhOut.fh {
|
||||
glog.V(4).Infof("server-side copy: skipped (same file handle)")
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
if fhIn.dirtyMetadata || fhOut.dirtyMetadata {
|
||||
glog.V(4).Infof("server-side copy: skipped (dirty metadata: in=%v out=%v)", fhIn.dirtyMetadata, fhOut.dirtyMetadata)
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
if in.OffIn != 0 || in.OffOut != 0 {
|
||||
glog.V(4).Infof("server-side copy: skipped (non-zero offsets: in=%d out=%d)", in.OffIn, in.OffOut)
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
srcEntry := fhIn.GetEntry()
|
||||
dstEntry := fhOut.GetEntry()
|
||||
if srcEntry == nil || dstEntry == nil {
|
||||
glog.V(4).Infof("server-side copy: skipped (nil entry: src=%v dst=%v)", srcEntry == nil, dstEntry == nil)
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
if srcEntry.IsDirectory || dstEntry.IsDirectory {
|
||||
glog.V(4).Infof("server-side copy: skipped (directory)")
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
srcPbEntry := srcEntry.GetEntry()
|
||||
dstPbEntry := dstEntry.GetEntry()
|
||||
if srcPbEntry == nil || dstPbEntry == nil || srcPbEntry.Attributes == nil || dstPbEntry.Attributes == nil {
|
||||
glog.V(4).Infof("server-side copy: skipped (missing entry attributes)")
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
sourceSize := int64(filer.FileSize(srcPbEntry))
|
||||
// go-fuse exposes CopyFileRange's return value as uint32, so the fast path
|
||||
// should only claim copies that can be reported without truncation.
|
||||
if sourceSize <= 0 || sourceSize > math.MaxUint32 || int64(in.Len) < sourceSize {
|
||||
glog.V(4).Infof("server-side copy: skipped (size mismatch: sourceSize=%d len=%d)", sourceSize, in.Len)
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
dstSize := int64(filer.FileSize(dstPbEntry))
|
||||
if dstSize != 0 || len(dstPbEntry.GetChunks()) > 0 || len(dstPbEntry.Content) > 0 {
|
||||
glog.V(4).Infof("server-side copy: skipped (destination not empty)")
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
srcPath := fhIn.FullPath()
|
||||
dstPath := fhOut.FullPath()
|
||||
if srcPath == "" || dstPath == "" || srcPath == dstPath {
|
||||
glog.V(4).Infof("server-side copy: skipped (invalid paths: src=%q dst=%q)", srcPath, dstPath)
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
if srcPbEntry.Attributes.Inode == 0 || dstPbEntry.Attributes.Inode == 0 {
|
||||
glog.V(4).Infof("server-side copy: skipped (missing inode preconditions: src=%d dst=%d)", srcPbEntry.Attributes.Inode, dstPbEntry.Attributes.Inode)
|
||||
return wholeFileServerCopyRequest{}, false
|
||||
}
|
||||
|
||||
return wholeFileServerCopyRequest{
|
||||
srcPath: srcPath,
|
||||
dstPath: dstPath,
|
||||
sourceSize: sourceSize,
|
||||
srcInode: srcPbEntry.Attributes.Inode,
|
||||
srcMtime: srcPbEntry.Attributes.Mtime,
|
||||
dstInode: dstPbEntry.Attributes.Inode,
|
||||
dstMtime: dstPbEntry.Attributes.Mtime,
|
||||
dstSize: dstSize,
|
||||
sourceMime: srcPbEntry.Attributes.Mime,
|
||||
sourceMd5: append([]byte(nil), srcPbEntry.Attributes.Md5...),
|
||||
copyRequestID: request_id.New(),
|
||||
}, true
|
||||
}
|
||||
|
||||
func (wfs *WFS) copyEntryViaFiler(cancel <-chan struct{}, copyRequest wholeFileServerCopyRequest) (*filer_pb.Entry, serverSideWholeFileCopyOutcome, error) {
|
||||
baseCtx, baseCancel := context.WithCancel(context.Background())
|
||||
defer baseCancel()
|
||||
|
||||
if cancel != nil {
|
||||
go func() {
|
||||
select {
|
||||
case <-cancel:
|
||||
baseCancel()
|
||||
case <-baseCtx.Done():
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
postCtx, postCancel := context.WithTimeout(baseCtx, filerCopyRequestTimeout)
|
||||
defer postCancel()
|
||||
|
||||
httpClient := util_http.GetGlobalHttpClient()
|
||||
if httpClient == nil {
|
||||
var err error
|
||||
httpClient, err = util_http.NewGlobalHttpClient()
|
||||
if err != nil {
|
||||
return nil, serverSideWholeFileCopyNotCommitted, fmt.Errorf("create filer copy http client: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
copyURL := &url.URL{
|
||||
Scheme: httpClient.GetHttpScheme(),
|
||||
Host: wfs.getCurrentFiler().ToHttpAddress(),
|
||||
Path: string(copyRequest.dstPath),
|
||||
}
|
||||
query := copyURL.Query()
|
||||
query.Set(filer.CopyQueryParamFrom, string(copyRequest.srcPath))
|
||||
query.Set(filer.CopyQueryParamOverwrite, "true")
|
||||
query.Set(filer.CopyQueryParamDataOnly, "true")
|
||||
query.Set(filer.CopyQueryParamRequestID, copyRequest.copyRequestID)
|
||||
query.Set(filer.CopyQueryParamSourceInode, fmt.Sprintf("%d", copyRequest.srcInode))
|
||||
query.Set(filer.CopyQueryParamSourceMtime, fmt.Sprintf("%d", copyRequest.srcMtime))
|
||||
query.Set(filer.CopyQueryParamSourceSize, fmt.Sprintf("%d", copyRequest.sourceSize))
|
||||
query.Set(filer.CopyQueryParamDestinationInode, fmt.Sprintf("%d", copyRequest.dstInode))
|
||||
query.Set(filer.CopyQueryParamDestinationMtime, fmt.Sprintf("%d", copyRequest.dstMtime))
|
||||
query.Set(filer.CopyQueryParamDestinationSize, fmt.Sprintf("%d", copyRequest.dstSize))
|
||||
copyURL.RawQuery = query.Encode()
|
||||
|
||||
req, err := http.NewRequestWithContext(postCtx, http.MethodPost, copyURL.String(), nil)
|
||||
if err != nil {
|
||||
return nil, serverSideWholeFileCopyNotCommitted, fmt.Errorf("create filer copy request: %w", err)
|
||||
}
|
||||
if jwt := wfs.filerCopyJWT(); jwt != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+string(jwt))
|
||||
}
|
||||
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return wfs.confirmServerSideWholeFileCopyAfterAmbiguousRequest(baseCtx, copyRequest, fmt.Errorf("execute filer copy request: %w", err))
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
return nil, serverSideWholeFileCopyNotCommitted, fmt.Errorf("filer copy %s => %s failed: status %d: %s", copyRequest.srcPath, copyRequest.dstPath, resp.StatusCode, string(body))
|
||||
}
|
||||
|
||||
readbackCtx, readbackCancel := context.WithTimeout(baseCtx, filerCopyReadbackTimeout)
|
||||
defer readbackCancel()
|
||||
|
||||
entry, err := filer_pb.GetEntry(readbackCtx, wfs, copyRequest.dstPath)
|
||||
if err != nil {
|
||||
return nil, serverSideWholeFileCopyCommitted, fmt.Errorf("reload copied entry %s: %w", copyRequest.dstPath, err)
|
||||
}
|
||||
if entry == nil {
|
||||
return nil, serverSideWholeFileCopyCommitted, fmt.Errorf("reload copied entry %s: not found", copyRequest.dstPath)
|
||||
}
|
||||
if entry.Attributes != nil && wfs.option != nil && wfs.option.UidGidMapper != nil {
|
||||
entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.FilerToLocal(entry.Attributes.Uid, entry.Attributes.Gid)
|
||||
}
|
||||
|
||||
return entry, serverSideWholeFileCopyCommitted, nil
|
||||
}
|
||||
|
||||
func (wfs *WFS) confirmServerSideWholeFileCopyAfterAmbiguousRequest(baseCtx context.Context, copyRequest wholeFileServerCopyRequest, requestErr error) (*filer_pb.Entry, serverSideWholeFileCopyOutcome, error) {
|
||||
readbackCtx, readbackCancel := context.WithTimeout(baseCtx, filerCopyReadbackTimeout)
|
||||
defer readbackCancel()
|
||||
|
||||
entry, err := filer_pb.GetEntry(readbackCtx, wfs, copyRequest.dstPath)
|
||||
if err == nil && entry != nil && entryMatchesServerSideWholeFileCopy(copyRequest, entry) {
|
||||
if entry.Attributes != nil && wfs.option != nil && wfs.option.UidGidMapper != nil {
|
||||
entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.FilerToLocal(entry.Attributes.Uid, entry.Attributes.Gid)
|
||||
}
|
||||
return entry, serverSideWholeFileCopyCommitted, nil
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, serverSideWholeFileCopyAmbiguous, fmt.Errorf("%w; post-copy readback failed: %v", requestErr, err)
|
||||
}
|
||||
if entry == nil {
|
||||
return nil, serverSideWholeFileCopyAmbiguous, fmt.Errorf("%w; destination %s was not readable after the ambiguous request", requestErr, copyRequest.dstPath)
|
||||
}
|
||||
return nil, serverSideWholeFileCopyAmbiguous, fmt.Errorf("%w; destination %s did not match the requested copy after the ambiguous request", requestErr, copyRequest.dstPath)
|
||||
}
|
||||
|
||||
func entryMatchesServerSideWholeFileCopy(copyRequest wholeFileServerCopyRequest, entry *filer_pb.Entry) bool {
|
||||
if entry == nil || entry.Attributes == nil {
|
||||
return false
|
||||
}
|
||||
if copyRequest.dstInode != 0 && entry.Attributes.Inode != copyRequest.dstInode {
|
||||
return false
|
||||
}
|
||||
if entry.Attributes.FileSize != uint64(copyRequest.sourceSize) {
|
||||
return false
|
||||
}
|
||||
if copyRequest.sourceMime != "" && entry.Attributes.Mime != copyRequest.sourceMime {
|
||||
return false
|
||||
}
|
||||
if len(copyRequest.sourceMd5) > 0 && !bytes.Equal(entry.Attributes.Md5, copyRequest.sourceMd5) {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (wfs *WFS) filerCopyJWT() security.EncodedJwt {
|
||||
if wfs.option == nil || len(wfs.option.FilerSigningKey) == 0 {
|
||||
return ""
|
||||
}
|
||||
return security.GenJwtForFilerServer(wfs.option.FilerSigningKey, wfs.option.FilerSigningExpiresAfterSec)
|
||||
}
|
||||
|
||||
func (wfs *WFS) invalidateCopyDestinationCache(inode uint64, fullPath util.FullPath) {
|
||||
if wfs.fuseServer != nil {
|
||||
if status := wfs.fuseServer.InodeNotify(inode, 0, -1); status != fuse.OK {
|
||||
glog.V(4).Infof("CopyFileRange invalidate inode %d: %v", inode, status)
|
||||
}
|
||||
dir, name := fullPath.DirAndName()
|
||||
if parentInode, found := wfs.inodeToPath.GetInode(util.FullPath(dir)); found {
|
||||
if status := wfs.fuseServer.EntryNotify(parentInode, name); status != fuse.OK {
|
||||
glog.V(4).Infof("CopyFileRange invalidate entry %s: %v", fullPath, status)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
355
weed/mount/weedfs_file_copy_range_test.go
Normal file
355
weed/mount/weedfs_file_copy_range_test.go
Normal file
@@ -0,0 +1,355 @@
|
||||
package mount
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/go-fuse/v2/fuse"
|
||||
"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
func TestWholeFileServerCopyCandidate(t *testing.T) {
|
||||
wfs := newCopyRangeTestWFS()
|
||||
|
||||
srcPath := util.FullPath("/src.txt")
|
||||
dstPath := util.FullPath("/dst.txt")
|
||||
srcInode := wfs.inodeToPath.Lookup(srcPath, 1, false, false, 0, true)
|
||||
dstInode := wfs.inodeToPath.Lookup(dstPath, 1, false, false, 0, true)
|
||||
|
||||
srcHandle := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{
|
||||
Name: "src.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
FileSize: 5,
|
||||
Inode: srcInode,
|
||||
},
|
||||
Content: []byte("hello"),
|
||||
})
|
||||
dstHandle := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{
|
||||
Name: "dst.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
Inode: dstInode,
|
||||
},
|
||||
})
|
||||
|
||||
srcHandle.RememberPath(srcPath)
|
||||
dstHandle.RememberPath(dstPath)
|
||||
|
||||
in := &fuse.CopyFileRangeIn{
|
||||
FhIn: uint64(srcHandle.fh),
|
||||
FhOut: uint64(dstHandle.fh),
|
||||
OffIn: 0,
|
||||
OffOut: 0,
|
||||
Len: 8,
|
||||
}
|
||||
|
||||
copyRequest, ok := wholeFileServerCopyCandidate(srcHandle, dstHandle, in)
|
||||
if !ok {
|
||||
t.Fatal("expected whole-file server copy candidate")
|
||||
}
|
||||
if copyRequest.srcPath != srcPath {
|
||||
t.Fatalf("source path = %q, want %q", copyRequest.srcPath, srcPath)
|
||||
}
|
||||
if copyRequest.dstPath != dstPath {
|
||||
t.Fatalf("destination path = %q, want %q", copyRequest.dstPath, dstPath)
|
||||
}
|
||||
if copyRequest.sourceSize != 5 {
|
||||
t.Fatalf("source size = %d, want 5", copyRequest.sourceSize)
|
||||
}
|
||||
if copyRequest.srcInode == 0 || copyRequest.dstInode == 0 {
|
||||
t.Fatalf("expected inode preconditions, got src=%d dst=%d", copyRequest.srcInode, copyRequest.dstInode)
|
||||
}
|
||||
|
||||
srcHandle.dirtyMetadata = true
|
||||
if _, ok := wholeFileServerCopyCandidate(srcHandle, dstHandle, in); ok {
|
||||
t.Fatal("dirty source handle should disable whole-file server copy")
|
||||
}
|
||||
srcHandle.dirtyMetadata = false
|
||||
|
||||
in.Len = 4
|
||||
if _, ok := wholeFileServerCopyCandidate(srcHandle, dstHandle, in); ok {
|
||||
t.Fatal("short copy request should disable whole-file server copy")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopyFileRangeUsesServerSideWholeFileCopy(t *testing.T) {
|
||||
wfs := newCopyRangeTestWFS()
|
||||
|
||||
srcPath := util.FullPath("/src.txt")
|
||||
dstPath := util.FullPath("/dst.txt")
|
||||
srcInode := wfs.inodeToPath.Lookup(srcPath, 1, false, false, 0, true)
|
||||
dstInode := wfs.inodeToPath.Lookup(dstPath, 1, false, false, 0, true)
|
||||
|
||||
srcHandle := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{
|
||||
Name: "src.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
FileSize: 5,
|
||||
Inode: srcInode,
|
||||
},
|
||||
Content: []byte("hello"),
|
||||
})
|
||||
dstHandle := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{
|
||||
Name: "dst.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
Inode: dstInode,
|
||||
},
|
||||
})
|
||||
|
||||
srcHandle.RememberPath(srcPath)
|
||||
dstHandle.RememberPath(dstPath)
|
||||
|
||||
originalCopy := performServerSideWholeFileCopy
|
||||
defer func() {
|
||||
performServerSideWholeFileCopy = originalCopy
|
||||
}()
|
||||
|
||||
var called bool
|
||||
performServerSideWholeFileCopy = func(cancel <-chan struct{}, gotWFS *WFS, copyRequest wholeFileServerCopyRequest) (*filer_pb.Entry, serverSideWholeFileCopyOutcome, error) {
|
||||
called = true
|
||||
if gotWFS != wfs {
|
||||
t.Fatalf("wfs = %p, want %p", gotWFS, wfs)
|
||||
}
|
||||
if copyRequest.srcPath != srcPath {
|
||||
t.Fatalf("source path = %q, want %q", copyRequest.srcPath, srcPath)
|
||||
}
|
||||
if copyRequest.dstPath != dstPath {
|
||||
t.Fatalf("destination path = %q, want %q", copyRequest.dstPath, dstPath)
|
||||
}
|
||||
return &filer_pb.Entry{
|
||||
Name: "dst.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
FileSize: 5,
|
||||
Mime: "text/plain; charset=utf-8",
|
||||
},
|
||||
Content: []byte("hello"),
|
||||
}, serverSideWholeFileCopyCommitted, nil
|
||||
}
|
||||
|
||||
written, status := wfs.CopyFileRange(make(chan struct{}), &fuse.CopyFileRangeIn{
|
||||
FhIn: uint64(srcHandle.fh),
|
||||
FhOut: uint64(dstHandle.fh),
|
||||
OffIn: 0,
|
||||
OffOut: 0,
|
||||
Len: 8,
|
||||
})
|
||||
if status != fuse.OK {
|
||||
t.Fatalf("CopyFileRange status = %v, want OK", status)
|
||||
}
|
||||
if written != 5 {
|
||||
t.Fatalf("CopyFileRange wrote %d bytes, want 5", written)
|
||||
}
|
||||
if !called {
|
||||
t.Fatal("expected server-side whole-file copy path to be used")
|
||||
}
|
||||
|
||||
gotEntry := dstHandle.GetEntry().GetEntry()
|
||||
if gotEntry.Attributes == nil || gotEntry.Attributes.FileSize != 5 {
|
||||
t.Fatalf("destination size = %v, want 5", gotEntry.GetAttributes().GetFileSize())
|
||||
}
|
||||
if string(gotEntry.Content) != "hello" {
|
||||
t.Fatalf("destination content = %q, want %q", string(gotEntry.Content), "hello")
|
||||
}
|
||||
if dstHandle.dirtyMetadata {
|
||||
t.Fatal("server-side whole-file copy should leave destination handle clean")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopyFileRangeDoesNotFallbackAfterCommittedServerCopyRefreshFailure(t *testing.T) {
|
||||
wfs := newCopyRangeTestWFSWithMetaCache(t)
|
||||
|
||||
srcPath := util.FullPath("/src.txt")
|
||||
dstPath := util.FullPath("/dst.txt")
|
||||
srcInode := wfs.inodeToPath.Lookup(srcPath, 1, false, false, 0, true)
|
||||
dstInode := wfs.inodeToPath.Lookup(dstPath, 1, false, false, 0, true)
|
||||
|
||||
srcHandle := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{
|
||||
Name: "src.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
FileSize: 5,
|
||||
Mime: "text/plain; charset=utf-8",
|
||||
Md5: []byte("abcde"),
|
||||
Inode: srcInode,
|
||||
},
|
||||
Content: []byte("hello"),
|
||||
})
|
||||
dstHandle := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{
|
||||
Name: "dst.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100600,
|
||||
Inode: dstInode,
|
||||
},
|
||||
})
|
||||
|
||||
srcHandle.RememberPath(srcPath)
|
||||
dstHandle.RememberPath(dstPath)
|
||||
|
||||
originalCopy := performServerSideWholeFileCopy
|
||||
defer func() {
|
||||
performServerSideWholeFileCopy = originalCopy
|
||||
}()
|
||||
|
||||
performServerSideWholeFileCopy = func(cancel <-chan struct{}, gotWFS *WFS, copyRequest wholeFileServerCopyRequest) (*filer_pb.Entry, serverSideWholeFileCopyOutcome, error) {
|
||||
if gotWFS != wfs || copyRequest.srcPath != srcPath || copyRequest.dstPath != dstPath {
|
||||
t.Fatalf("unexpected server-side copy call: wfs=%p src=%q dst=%q", gotWFS, copyRequest.srcPath, copyRequest.dstPath)
|
||||
}
|
||||
return nil, serverSideWholeFileCopyCommitted, errors.New("reload copied entry: transient filer read failure")
|
||||
}
|
||||
|
||||
written, status := wfs.CopyFileRange(make(chan struct{}), &fuse.CopyFileRangeIn{
|
||||
FhIn: uint64(srcHandle.fh),
|
||||
FhOut: uint64(dstHandle.fh),
|
||||
OffIn: 0,
|
||||
OffOut: 0,
|
||||
Len: 8,
|
||||
})
|
||||
if status != fuse.OK {
|
||||
t.Fatalf("CopyFileRange status = %v, want OK", status)
|
||||
}
|
||||
if written != 5 {
|
||||
t.Fatalf("CopyFileRange wrote %d bytes, want 5", written)
|
||||
}
|
||||
if dstHandle.dirtyMetadata {
|
||||
t.Fatal("committed server-side copy should not fall back to dirty-page copy")
|
||||
}
|
||||
|
||||
gotEntry := dstHandle.GetEntry().GetEntry()
|
||||
if gotEntry.GetAttributes().GetFileSize() != 5 {
|
||||
t.Fatalf("destination size = %d, want 5", gotEntry.GetAttributes().GetFileSize())
|
||||
}
|
||||
if gotEntry.GetAttributes().GetFileMode() != 0100600 {
|
||||
t.Fatalf("destination mode = %#o, want %#o", gotEntry.GetAttributes().GetFileMode(), uint32(0100600))
|
||||
}
|
||||
if string(gotEntry.GetContent()) != "hello" {
|
||||
t.Fatalf("destination content = %q, want %q", string(gotEntry.GetContent()), "hello")
|
||||
}
|
||||
|
||||
cachedEntry, err := wfs.metaCache.FindEntry(context.Background(), dstPath)
|
||||
if err != nil {
|
||||
t.Fatalf("metaCache find entry: %v", err)
|
||||
}
|
||||
if cachedEntry.FileSize != 5 {
|
||||
t.Fatalf("metaCache destination size = %d, want 5", cachedEntry.FileSize)
|
||||
}
|
||||
if cachedEntry.Mime != "text/plain; charset=utf-8" {
|
||||
t.Fatalf("metaCache destination mime = %q, want %q", cachedEntry.Mime, "text/plain; charset=utf-8")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCopyFileRangeReturnsEIOForAmbiguousServerSideCopy(t *testing.T) {
|
||||
wfs := newCopyRangeTestWFS()
|
||||
|
||||
srcPath := util.FullPath("/src.txt")
|
||||
dstPath := util.FullPath("/dst.txt")
|
||||
srcInode := wfs.inodeToPath.Lookup(srcPath, 1, false, false, 0, true)
|
||||
dstInode := wfs.inodeToPath.Lookup(dstPath, 1, false, false, 0, true)
|
||||
|
||||
srcHandle := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{
|
||||
Name: "src.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100644,
|
||||
FileSize: 5,
|
||||
Inode: srcInode,
|
||||
},
|
||||
Content: []byte("hello"),
|
||||
})
|
||||
dstHandle := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{
|
||||
Name: "dst.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
FileMode: 0100600,
|
||||
Inode: dstInode,
|
||||
},
|
||||
})
|
||||
|
||||
srcHandle.RememberPath(srcPath)
|
||||
dstHandle.RememberPath(dstPath)
|
||||
|
||||
originalCopy := performServerSideWholeFileCopy
|
||||
defer func() {
|
||||
performServerSideWholeFileCopy = originalCopy
|
||||
}()
|
||||
|
||||
performServerSideWholeFileCopy = func(cancel <-chan struct{}, gotWFS *WFS, copyRequest wholeFileServerCopyRequest) (*filer_pb.Entry, serverSideWholeFileCopyOutcome, error) {
|
||||
if gotWFS != wfs || copyRequest.srcPath != srcPath || copyRequest.dstPath != dstPath {
|
||||
t.Fatalf("unexpected server-side copy call: wfs=%p src=%q dst=%q", gotWFS, copyRequest.srcPath, copyRequest.dstPath)
|
||||
}
|
||||
return nil, serverSideWholeFileCopyAmbiguous, errors.New("transport timeout after request dispatch")
|
||||
}
|
||||
|
||||
written, status := wfs.CopyFileRange(make(chan struct{}), &fuse.CopyFileRangeIn{
|
||||
FhIn: uint64(srcHandle.fh),
|
||||
FhOut: uint64(dstHandle.fh),
|
||||
OffIn: 0,
|
||||
OffOut: 0,
|
||||
Len: 8,
|
||||
})
|
||||
if status != fuse.EIO {
|
||||
t.Fatalf("CopyFileRange status = %v, want EIO", status)
|
||||
}
|
||||
if written != 0 {
|
||||
t.Fatalf("CopyFileRange wrote %d bytes, want 0", written)
|
||||
}
|
||||
if dstHandle.dirtyMetadata {
|
||||
t.Fatal("ambiguous server-side copy should not fall back to dirty-page copy")
|
||||
}
|
||||
if dstHandle.GetEntry().GetEntry().GetAttributes().GetFileSize() != 0 {
|
||||
t.Fatalf("destination size = %d, want 0", dstHandle.GetEntry().GetEntry().GetAttributes().GetFileSize())
|
||||
}
|
||||
}
|
||||
|
||||
func newCopyRangeTestWFS() *WFS {
|
||||
wfs := &WFS{
|
||||
option: &Option{
|
||||
ChunkSizeLimit: 1024,
|
||||
ConcurrentReaders: 1,
|
||||
VolumeServerAccess: "filerProxy",
|
||||
FilerAddresses: []pb.ServerAddress{"127.0.0.1:8888"},
|
||||
},
|
||||
inodeToPath: NewInodeToPath(util.FullPath("/"), 0),
|
||||
fhMap: NewFileHandleToInode(),
|
||||
fhLockTable: util.NewLockTable[FileHandleId](),
|
||||
}
|
||||
wfs.copyBufferPool.New = func() any {
|
||||
return make([]byte, 1024)
|
||||
}
|
||||
return wfs
|
||||
}
|
||||
|
||||
func newCopyRangeTestWFSWithMetaCache(t *testing.T) *WFS {
|
||||
t.Helper()
|
||||
|
||||
wfs := newCopyRangeTestWFS()
|
||||
root := util.FullPath("/")
|
||||
wfs.inodeToPath.MarkChildrenCached(root)
|
||||
uidGidMapper, err := meta_cache.NewUidGidMapper("", "")
|
||||
if err != nil {
|
||||
t.Fatalf("create uid/gid mapper: %v", err)
|
||||
}
|
||||
wfs.metaCache = meta_cache.NewMetaCache(
|
||||
filepath.Join(t.TempDir(), "meta"),
|
||||
uidGidMapper,
|
||||
root,
|
||||
func(path util.FullPath) {
|
||||
wfs.inodeToPath.MarkChildrenCached(path)
|
||||
},
|
||||
func(path util.FullPath) bool {
|
||||
return wfs.inodeToPath.IsChildrenCached(path)
|
||||
},
|
||||
func(util.FullPath, *filer_pb.Entry) {},
|
||||
nil,
|
||||
)
|
||||
t.Cleanup(func() {
|
||||
wfs.metaCache.Shutdown()
|
||||
})
|
||||
|
||||
return wfs
|
||||
}
|
||||
Reference in New Issue
Block a user