seaweedfs/weed/filesys/filehandle.go

239 lines
6.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"
"math"
2019-05-03 07:24:35 +00:00
"mime"
"path"
"time"
"github.com/gabriel-vasile/mimetype"
2018-05-23 10:08:46 +00:00
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
2018-05-27 18:52:26 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
2018-05-23 10:08:46 +00:00
)
type FileHandle struct {
// cache file has been written to
dirtyPages *ContinuousDirtyPages
contentType string
dirtyMetadata bool
handle uint64
2018-06-06 09:09:57 +00:00
2018-05-25 07:57:25 +00:00
f *File
RequestId fuse.RequestID // unique ID for request
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
2018-05-23 10:08:46 +00:00
}
2018-06-06 09:09:57 +00:00
func newFileHandle(file *File, uid, gid uint32) *FileHandle {
fh := &FileHandle{
2018-06-06 09:09:57 +00:00
f: file,
dirtyPages: newDirtyPages(file),
Uid: uid,
Gid: gid,
}
if fh.f.entry != nil {
fh.f.entry.Attributes.FileSize = filer2.TotalSize(fh.f.entry.Chunks)
}
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
2018-06-06 09:09:57 +00:00
glog.V(4).Infof("%s read fh %d: [%d,%d)", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(req.Size))
2018-05-23 10:08:46 +00:00
buff := make([]byte, req.Size)
2020-02-26 06:23:59 +00:00
totalRead, err := fh.readFromChunks(buff, req.Offset)
if err == nil {
2020-02-26 06:23:59 +00:00
dirtyOffset, dirtySize := fh.readFromDirtyPages(buff, req.Offset)
if totalRead+req.Offset < dirtyOffset+int64(dirtySize) {
totalRead = dirtyOffset + int64(dirtySize) - req.Offset
}
}
resp.Data = buff[:totalRead]
if err != nil {
glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
2020-01-24 09:41:31 +00:00
return fuse.EIO
}
return err
}
2020-02-26 06:23:59 +00:00
func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (offset int64, size int) {
return fh.dirtyPages.ReadDirtyData(buff, startOffset)
}
2020-02-26 06:23:59 +00:00
func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
2018-06-06 09:09:57 +00:00
// this value should come from the filer instead of the old f
if len(fh.f.entry.Chunks) == 0 {
2020-01-25 15:34:09 +00:00
glog.V(1).Infof("empty fh %v", fh.f.fullpath())
return 0, nil
2018-05-23 10:08:46 +00:00
}
if fh.f.entryViewCache == nil {
fh.f.entryViewCache = filer2.NonOverlappingVisibleIntervals(fh.f.entry.Chunks)
fh.f.reader = nil
}
if fh.f.reader == nil {
chunkViews := filer2.ViewFromVisibleIntervals(fh.f.entryViewCache, 0, math.MaxInt32)
fh.f.reader = filer2.NewChunkReaderAtFromClient(fh.f.wfs, chunkViews, fh.f.wfs.chunkCache)
}
totalRead, err := fh.f.reader.ReadAt(buff, offset)
2019-06-21 06:45:30 +00:00
if err != nil {
glog.Errorf("file handle read %s: %v", fh.f.fullpath(), err)
}
// glog.V(0).Infof("file handle read %s [%d,%d] %d : %v", fh.f.fullpath(), 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 {
2018-05-23 10:08:46 +00:00
// write the request to volume servers
data := make([]byte, len(req.Data))
copy(data, req.Data)
fh.f.entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(fh.f.entry.Attributes.FileSize)))
2020-01-27 08:54:21 +00:00
// glog.V(0).Infof("%v write [%d,%d)", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)))
chunks, err := fh.dirtyPages.AddPage(req.Offset, data)
2018-05-23 10:08:46 +00:00
if err != nil {
glog.Errorf("%v write fh %d: [%d,%d): %v", fh.f.fullpath(), fh.handle, req.Offset, req.Offset+int64(len(data)), err)
2020-01-24 09:41:31 +00:00
return fuse.EIO
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
detectedMIME := mimetype.Detect(data)
2019-12-17 09:47:12 +00:00
fh.contentType = detectedMIME.String()
if ext := path.Ext(fh.f.Name); ext != detectedMIME.Extension() {
2019-03-27 21:25:18 +00:00
fh.contentType = mime.TypeByExtension(ext)
}
fh.dirtyMetadata = true
}
2019-01-05 23:16:39 +00:00
if len(chunks) > 0 {
2019-12-14 19:04:20 +00:00
fh.f.addChunks(chunks)
fh.dirtyMetadata = true
}
2018-05-23 10:08:46 +00:00
return nil
}
func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
2018-06-06 09:09:57 +00:00
glog.V(4).Infof("%v release fh %d", fh.f.fullpath(), fh.handle)
fh.f.isOpen--
2018-12-28 11:27:48 +00:00
if fh.f.isOpen <= 0 {
fh.dirtyPages.releaseResource()
fh.f.wfs.ReleaseHandle(fh.f.fullpath(), fuse.HandleID(fh.handle))
}
fh.f.entryViewCache = nil
fh.f.reader = nil
2018-05-23 10:08:46 +00:00
return nil
}
func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
// fflush works at fh level
// send the data to the OS
2018-06-06 09:09:57 +00:00
glog.V(4).Infof("%s fh %d flush %v", fh.f.fullpath(), fh.handle, req)
2018-05-23 10:08:46 +00:00
chunks, err := fh.dirtyPages.FlushToStorage()
if err != nil {
2020-01-25 15:34:09 +00:00
glog.Errorf("flush %s: %v", fh.f.fullpath(), err)
2020-01-24 09:41:31 +00:00
return fuse.EIO
}
2019-01-05 23:16:39 +00:00
if len(chunks) > 0 {
fh.f.addChunks(chunks)
fh.dirtyMetadata = true
}
if !fh.dirtyMetadata {
2018-05-23 10:08:46 +00:00
return nil
}
err = fh.f.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
2018-05-23 10:08:46 +00:00
if fh.f.entry.Attributes != nil {
fh.f.entry.Attributes.Mime = fh.contentType
fh.f.entry.Attributes.Uid = req.Uid
fh.f.entry.Attributes.Gid = req.Gid
fh.f.entry.Attributes.Mtime = time.Now().Unix()
fh.f.entry.Attributes.Crtime = time.Now().Unix()
fh.f.entry.Attributes.FileMode = uint32(0666 &^ fh.f.wfs.option.Umask)
fh.f.entry.Attributes.Collection = fh.dirtyPages.collection
fh.f.entry.Attributes.Replication = fh.dirtyPages.replication
}
request := &filer_pb.CreateEntryRequest{
Directory: fh.f.dir.FullPath(),
Entry: fh.f.entry,
2018-05-23 10:08:46 +00:00
}
2020-01-25 15:34:09 +00:00
glog.V(3).Infof("%s set chunks: %v", fh.f.fullpath(), len(fh.f.entry.Chunks))
2019-06-21 19:14:40 +00:00
for i, chunk := range fh.f.entry.Chunks {
2020-01-24 18:02:53 +00:00
glog.V(3).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
2019-06-21 19:14:40 +00:00
}
2019-01-01 10:33:57 +00:00
chunks, garbages := filer2.CompactFileChunks(fh.f.entry.Chunks)
fh.f.entry.Chunks = chunks
2019-01-05 23:16:39 +00:00
// fh.f.entryViewCache = nil
2019-01-01 10:33:57 +00:00
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
}
fh.f.wfs.deleteFileChunks(garbages)
2019-06-21 19:14:40 +00:00
for i, chunk := range garbages {
2020-01-25 15:34:09 +00:00
glog.V(3).Infof("garbage %s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.FileId, chunk.Offset, chunk.Offset+int64(chunk.Size))
2019-06-21 19:14:40 +00:00
}
2018-05-23 10:08:46 +00:00
return nil
})
if err == nil {
fh.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
}