seaweedfs/weed/filesys/file.go

204 lines
4.9 KiB
Go
Raw Normal View History

2018-05-06 05:47:16 +00:00
package filesys
import (
2018-05-27 18:52:26 +00:00
"context"
"os"
"path/filepath"
"sort"
"time"
2018-05-27 18:52:26 +00:00
"github.com/chrislusf/seaweedfs/weed/filer2"
2018-05-08 08:59:43 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
2018-05-10 06:18:02 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
2018-05-06 05:47:16 +00:00
)
const blockSize = 512
2018-05-08 08:59:43 +00:00
var _ = fs.Node(&File{})
var _ = fs.NodeOpener(&File{})
var _ = fs.NodeFsyncer(&File{})
var _ = fs.NodeSetattrer(&File{})
2018-05-08 08:59:43 +00:00
2018-05-06 05:47:16 +00:00
type File struct {
Name string
dir *Dir
wfs *WFS
entry *filer_pb.Entry
2018-12-31 23:10:14 +00:00
entryViewCache []filer2.VisibleInterval
isOpen bool
2018-05-06 05:47:16 +00:00
}
2018-06-06 09:09:57 +00:00
func (file *File) fullpath() string {
return filepath.Join(file.dir.Path, file.Name)
}
2018-06-06 09:09:57 +00:00
func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error {
if err := file.maybeLoadAttributes(ctx); err != nil {
return err
}
2018-05-08 08:59:43 +00:00
attr.Mode = os.FileMode(file.entry.Attributes.FileMode)
attr.Size = filer2.TotalSize(file.entry.Chunks)
attr.Mtime = time.Unix(file.entry.Attributes.Mtime, 0)
attr.Gid = file.entry.Attributes.Gid
attr.Uid = file.entry.Attributes.Uid
attr.Blocks = attr.Size/blockSize + 1
2018-10-14 07:18:52 +00:00
attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
2018-05-19 20:51:44 +00:00
return nil
2018-05-08 08:59:43 +00:00
}
func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
2018-05-23 10:08:46 +00:00
2018-06-06 09:09:57 +00:00
glog.V(3).Infof("%v file open %+v", file.fullpath(), req)
file.isOpen = true
2018-06-06 09:09:57 +00:00
handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid)
resp.Handle = fuse.HandleID(handle.handle)
glog.V(3).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
return handle, nil
}
func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
if err := file.maybeLoadAttributes(ctx); err != nil {
return err
}
glog.V(3).Infof("%v file setattr %+v, old:%+v", file.fullpath(), req, file.entry.Attributes)
if req.Valid.Size() {
2018-06-06 09:09:57 +00:00
glog.V(3).Infof("%v file setattr set size=%v", file.fullpath(), req.Size)
if req.Size == 0 {
2018-05-25 06:20:12 +00:00
// fmt.Printf("truncate %v \n", fullPath)
file.entry.Chunks = nil
file.entryViewCache = nil
}
file.entry.Attributes.FileSize = req.Size
}
if req.Valid.Mode() {
file.entry.Attributes.FileMode = uint32(req.Mode)
}
if req.Valid.Uid() {
file.entry.Attributes.Uid = req.Uid
}
if req.Valid.Gid() {
file.entry.Attributes.Gid = req.Gid
}
2018-12-29 06:37:18 +00:00
if req.Valid.Crtime() {
file.entry.Attributes.Crtime = req.Crtime.Unix()
}
if req.Valid.Mtime() {
file.entry.Attributes.Mtime = req.Mtime.Unix()
}
2019-02-15 18:00:27 +00:00
if file.isOpen {
return nil
}
return file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.UpdateEntryRequest{
Directory: file.dir.Path,
2018-09-22 07:12:10 +00:00
Entry: file.entry,
}
glog.V(1).Infof("set attr file entry: %v", request)
_, err := client.UpdateEntry(ctx, request)
if err != nil {
glog.V(0).Infof("UpdateEntry file %s/%s: %v", file.dir.Path, file.Name, err)
return fuse.EIO
}
return nil
})
}
func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
// fsync works at OS level
2018-06-06 06:37:41 +00:00
// write the file chunks to the filerGrpcAddress
glog.V(3).Infof("%s/%s fsync file %+v", file.dir.Path, file.Name, req)
2018-05-19 20:51:44 +00:00
2018-05-08 08:59:43 +00:00
return nil
2018-05-06 05:47:16 +00:00
}
func (file *File) maybeLoadAttributes(ctx context.Context) error {
if file.entry == nil || !file.isOpen {
item := file.wfs.listDirectoryEntriesCache.Get(file.fullpath())
if item != nil && !item.Expired() {
entry := item.Value().(*filer_pb.Entry)
2019-01-05 23:21:56 +00:00
file.setEntry(entry)
// glog.V(1).Infof("file attr read cached %v attributes", file.Name)
} else {
err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.LookupDirectoryEntryRequest{
Name: file.Name,
Directory: file.dir.Path,
}
resp, err := client.LookupDirectoryEntry(ctx, request)
if err != nil {
2018-12-28 23:10:23 +00:00
glog.V(3).Infof("file attr read file %v: %v", request, err)
return fuse.ENOENT
}
2019-01-05 23:21:56 +00:00
file.setEntry(resp.Entry)
2018-12-28 23:10:23 +00:00
glog.V(3).Infof("file attr %v %+v: %d", file.fullpath(), file.entry.Attributes, filer2.TotalSize(file.entry.Chunks))
2019-01-04 06:21:39 +00:00
// file.wfs.listDirectoryEntriesCache.Set(file.fullpath(), file.entry, file.wfs.option.EntryCacheTtl)
2018-11-07 19:35:13 +00:00
return nil
})
if err != nil {
return err
}
}
}
return nil
}
2019-01-05 23:16:39 +00:00
func (file *File) addChunk(chunk *filer_pb.FileChunk) {
if chunk != nil {
file.addChunks([]*filer_pb.FileChunk{chunk})
}
}
func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
sort.Slice(chunks, func(i, j int) bool {
return chunks[i].Mtime < chunks[j].Mtime
})
var newVisibles []filer2.VisibleInterval
2019-01-05 23:16:39 +00:00
for _, chunk := range chunks {
newVisibles = filer2.MergeIntoVisibles(file.entryViewCache, newVisibles, chunk)
t := file.entryViewCache[:0]
file.entryViewCache = newVisibles
newVisibles = t
2019-01-05 23:16:39 +00:00
}
file.entry.Chunks = append(file.entry.Chunks, chunks...)
2019-01-05 23:16:39 +00:00
}
2019-01-05 23:21:56 +00:00
func (file *File) setEntry(entry *filer_pb.Entry) {
file.entry = entry
file.entryViewCache = filer2.NonOverlappingVisibleIntervals(file.entry.Chunks)
}