seaweedfs/weed/filesys/filehandle.go

348 lines
9.4 KiB
Go
Raw Normal View History

2018-05-23 10:08:46 +00:00
package filesys
import (
2018-05-27 18:52:26 +00:00
"context"
2018-05-23 10:08:46 +00:00
"fmt"
2020-06-10 01:04:40 +00:00
"io"
"math"
2020-04-14 18:32:31 +00:00
"net/http"
"os"
"sync"
2019-05-03 07:24:35 +00:00
"time"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
2020-09-01 07:21:19 +00:00
"github.com/chrislusf/seaweedfs/weed/filer"
2018-05-23 10:08:46 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
2018-05-27 18:52:26 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
2018-05-23 10:08:46 +00:00
)
type FileHandle struct {
// cache file has been written to
dirtyPages *PageWriter
entryViewCache []filer.VisibleInterval
reader io.ReaderAt
contentType string
handle uint64
2021-03-23 05:32:47 +00:00
sync.Mutex
2018-06-06 09:09:57 +00:00
2018-05-25 07:57:25 +00:00
f *File
2022-01-17 11:19:11 +00:00
NodeId fuse.NodeID // file or directory the request is about
Uid uint32 // user ID of process making request
Gid uint32 // group ID of process making request
isDeleted bool
2018-05-23 10:08:46 +00:00
}
2021-12-20 09:11:43 +00:00
func newFileHandle(file *File, uid, gid uint32) *FileHandle {
fh := &FileHandle{
2022-01-17 21:53:30 +00:00
f: file,
Uid: uid,
Gid: gid,
2018-06-06 09:09:57 +00:00
}
2022-01-17 21:53:30 +00:00
// dirtyPages: newContinuousDirtyPages(file, writeOnly),
fh.dirtyPages = newPageWriter(fh, file.wfs.option.ChunkSizeLimit)
entry := fh.f.getEntry()
if entry != nil {
entry.Attributes.FileSize = filer.FileSize(entry)
}
return fh
2018-06-06 09:09:57 +00:00
}
2018-05-23 10:08:46 +00:00
var _ = fs.Handle(&FileHandle{})
2018-05-27 18:52:26 +00:00
// var _ = fs.HandleReadAller(&FileHandle{})
var _ = fs.HandleReader(&FileHandle{})
2018-05-23 10:08:46 +00:00
var _ = fs.HandleFlusher(&FileHandle{})
var _ = fs.HandleWriter(&FileHandle{})
var _ = fs.HandleReleaser(&FileHandle{})
func (fh *FileHandle) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
2018-05-23 10:08:46 +00:00
2021-03-23 05:32:47 +00:00
fh.Lock()
defer fh.Unlock()
glog.V(4).Infof("%s read fh %d: [%d,%d) size %d resp.Data cap=%d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, cap(resp.Data))
if req.Size <= 0 {
return nil
}
2018-05-23 10:08:46 +00:00
2020-08-16 23:24:40 +00:00
buff := resp.Data[:cap(resp.Data)]
if req.Size > cap(resp.Data) {
// should not happen
buff = make([]byte, req.Size)
}
fh.lockForRead(req.Offset, len(buff))
defer fh.unlockForRead(req.Offset, len(buff))
2020-02-26 06:23:59 +00:00
totalRead, err := fh.readFromChunks(buff, req.Offset)
2021-01-18 09:14:42 +00:00
if err == nil || err == io.EOF {
maxStop := fh.readFromDirtyPages(buff, req.Offset)
totalRead = max(maxStop-req.Offset, totalRead)
}
if err == io.EOF {
err = nil
}
if err != nil {
2020-08-31 04:01:44 +00:00
glog.Warningf("file handle read %s %d: %v", fh.f.fullpath(), totalRead, err)
2020-10-10 22:43:22 +00:00
return fuse.EIO
}
2020-08-17 23:04:56 +00:00
if totalRead > int64(len(buff)) {
glog.Warningf("%s FileHandle Read %d: [%d,%d) size %d totalRead %d", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size), req.Size, totalRead)
totalRead = min(int64(len(buff)), totalRead)
}
if err == nil {
resp.Data = buff[:totalRead]
}
2020-08-17 06:49:10 +00:00
return err
}
func (fh *FileHandle) lockForRead(startOffset int64, size int) {
fh.dirtyPages.LockForRead(startOffset, startOffset+int64(size))
}
func (fh *FileHandle) unlockForRead(startOffset int64, size int) {
fh.dirtyPages.UnlockForRead(startOffset, startOffset+int64(size))
}
func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (maxStop int64) {
maxStop = fh.dirtyPages.ReadDirtyDataAt(buff, startOffset)
return
}
2020-02-26 06:23:59 +00:00
func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
entry := fh.f.getEntry()
if entry == nil {
return 0, io.EOF
}
2021-08-10 05:11:57 +00:00
if entry.IsInRemoteOnly() {
glog.V(4).Infof("download remote entry %s", fh.f.fullpath())
newEntry, err := fh.f.downloadRemoteEntry(entry)
if err != nil {
glog.V(1).Infof("download remote entry %s: %v", fh.f.fullpath(), err)
return 0, err
}
entry = newEntry
}
fileSize := int64(filer.FileSize(entry))
fileFullPath := fh.f.fullpath()
2020-08-16 07:49:08 +00:00
if fileSize == 0 {
glog.V(1).Infof("empty fh %v", fileFullPath)
2020-08-16 07:49:08 +00:00
return 0, io.EOF
2018-05-23 10:08:46 +00:00
}
if offset+int64(len(buff)) <= int64(len(entry.Content)) {
totalRead := copy(buff, entry.Content[offset:])
glog.V(4).Infof("file handle read cached %s [%d,%d] %d", fileFullPath, offset, offset+int64(totalRead), totalRead)
2020-11-30 12:34:04 +00:00
return int64(totalRead), nil
}
var chunkResolveErr error
2022-01-18 07:02:30 +00:00
if fh.entryViewCache == nil {
fh.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks, 0, math.MaxInt64)
if chunkResolveErr != nil {
return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
}
fh.reader = nil
}
reader := fh.reader
if reader == nil {
chunkViews := filer.ViewFromVisibleIntervals(fh.entryViewCache, 0, math.MaxInt64)
2022-01-22 20:35:09 +00:00
glog.V(4).Infof("file handle read %s [%d,%d) from %d views", fileFullPath, offset, offset+int64(len(buff)), len(chunkViews))
2022-01-17 22:02:37 +00:00
for _, chunkView := range chunkViews {
glog.V(4).Infof(" read %s [%d,%d) from chunk %+v", fileFullPath, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.FileId)
}
reader = filer.NewChunkReaderAtFromClient(fh.f.wfs.LookupFn(), chunkViews, fh.f.wfs.chunkCache, fileSize)
}
fh.reader = reader
totalRead, err := reader.ReadAt(buff, offset)
2020-10-25 03:12:04 +00:00
if err != nil && err != io.EOF {
glog.Errorf("file handle read %s: %v", fileFullPath, err)
2019-06-21 06:45:30 +00:00
}
2022-01-18 04:41:00 +00:00
glog.V(4).Infof("file handle read %s [%d,%d] %d : %v", fileFullPath, offset, offset+int64(totalRead), totalRead, err)
return int64(totalRead), err
2018-05-23 10:08:46 +00:00
}
// Write to the file handle
func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
fh.dirtyPages.writerPattern.MonitorWriteAt(req.Offset, len(req.Data))
fh.Lock()
defer fh.Unlock()
2018-05-23 10:08:46 +00:00
// write the request to volume servers
2020-10-21 05:54:21 +00:00
data := req.Data
if len(data) <= 512 && req.Offset == 0 {
// fuse message cacheable size
data = make([]byte, len(req.Data))
copy(data, req.Data)
}
entry := fh.f.getEntry()
if entry == nil {
return fuse.EIO
}
entry.Content = nil
entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(entry.Attributes.FileSize)))
2021-05-11 04:47:51 +00:00
// glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
2020-10-15 06:28:03 +00:00
fh.dirtyPages.AddPage(req.Offset, data)
2018-05-23 10:08:46 +00:00
resp.Size = len(data)
2018-05-23 10:08:46 +00:00
if req.Offset == 0 {
2019-03-27 21:25:18 +00:00
// detect mime type
2020-04-14 18:32:31 +00:00
fh.contentType = http.DetectContentType(data)
fh.f.dirtyMetadata = true
}
2020-10-15 06:28:03 +00:00
fh.f.dirtyMetadata = true
2018-05-23 10:08:46 +00:00
return nil
}
func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
2021-04-17 17:48:22 +00:00
glog.V(4).Infof("Release %v fh %d open=%d", fh.f.fullpath(), fh.handle, fh.f.isOpen)
2018-06-06 09:09:57 +00:00
2021-08-11 13:58:35 +00:00
fh.f.wfs.handlesLock.Lock()
2021-04-17 17:48:22 +00:00
fh.f.isOpen--
2021-08-11 13:58:35 +00:00
fh.f.wfs.handlesLock.Unlock()
2021-04-17 17:48:22 +00:00
2021-04-21 02:56:51 +00:00
if fh.f.isOpen <= 0 {
fh.f.entry = nil
fh.entryViewCache = nil
fh.reader = nil
fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
2021-12-24 01:17:32 +00:00
fh.dirtyPages.Destroy()
}
2021-04-21 02:56:51 +00:00
if fh.f.isOpen < 0 {
glog.V(0).Infof("Release reset %s open count %d => %d", fh.f.Name, fh.f.isOpen, 0)
fh.f.isOpen = 0
return nil
}
2018-05-23 10:08:46 +00:00
return nil
}
func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
glog.V(4).Infof("Flush %v fh %d", fh.f.fullpath(), fh.handle)
if fh.isDeleted {
glog.V(4).Infof("Flush %v fh %d skip deleted", fh.f.fullpath(), fh.handle)
return nil
}
fh.Lock()
defer fh.Unlock()
if err := fh.doFlush(ctx, req.Header); err != nil {
glog.Errorf("Flush doFlush %s: %v", fh.f.Name, err)
return err
}
return nil
2020-08-19 08:27:10 +00:00
}
func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
2020-10-15 06:28:03 +00:00
// flush works at fh level
2018-05-23 10:08:46 +00:00
// send the data to the OS
2020-08-29 18:56:22 +00:00
glog.V(4).Infof("doFlush %s fh %d", fh.f.fullpath(), fh.handle)
2018-05-23 10:08:46 +00:00
2021-05-09 22:15:18 +00:00
if err := fh.dirtyPages.FlushData(); err != nil {
glog.Errorf("%v doFlush: %v", fh.f.fullpath(), err)
return fuse.EIO
}
if !fh.f.dirtyMetadata {
2018-05-23 10:08:46 +00:00
return nil
}
err := fh.f.wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
2018-05-23 10:08:46 +00:00
entry := fh.f.getEntry()
if entry == nil {
return nil
}
if entry.Attributes != nil {
entry.Attributes.Mime = fh.contentType
if entry.Attributes.Uid == 0 {
entry.Attributes.Uid = header.Uid
}
if entry.Attributes.Gid == 0 {
entry.Attributes.Gid = header.Gid
}
if entry.Attributes.Crtime == 0 {
entry.Attributes.Crtime = time.Now().Unix()
}
entry.Attributes.Mtime = time.Now().Unix()
entry.Attributes.FileMode = uint32(os.FileMode(entry.Attributes.FileMode) &^ fh.f.wfs.option.Umask)
2021-05-09 22:28:54 +00:00
entry.Attributes.Collection, entry.Attributes.Replication = fh.dirtyPages.GetStorageOptions()
}
request := &filer_pb.CreateEntryRequest{
Directory: fh.f.dir.FullPath(),
Entry: entry,
Signatures: []int32{fh.f.wfs.signature},
2018-05-23 10:08:46 +00:00
}
glog.V(4).Infof("%s set chunks: %v", fh.f.fullpath(), len(entry.Chunks))
for i, chunk := range entry.Chunks {
2020-08-16 02:55:28 +00:00
glog.V(4).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.GetFileIdString(), chunk.Offset, chunk.Offset+int64(chunk.Size))
2019-06-21 19:14:40 +00:00
}
2019-01-01 10:33:57 +00:00
manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.Chunks)
2020-08-23 23:59:01 +00:00
chunks, _ := filer.CompactFileChunks(fh.f.wfs.LookupFn(), nonManifestChunks)
2021-12-20 09:11:43 +00:00
chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
if manifestErr != nil {
// not good, but should be ok
glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
}
entry.Chunks = append(chunks, manifestChunks...)
fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)
if err := filer_pb.CreateEntry(client, request); err != nil {
glog.Errorf("fh flush create %s: %v", fh.f.fullpath(), err)
return fmt.Errorf("fh flush create %s: %v", fh.f.fullpath(), err)
2018-05-23 10:08:46 +00:00
}
2020-09-01 07:21:19 +00:00
fh.f.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
2018-05-23 10:08:46 +00:00
return nil
})
if err == nil {
fh.f.dirtyMetadata = false
}
2020-01-24 09:41:31 +00:00
if err != nil {
glog.Errorf("%v fh %d flush: %v", fh.f.fullpath(), fh.handle, err)
return fuse.EIO
}
return nil
2018-05-23 10:08:46 +00:00
}