seaweedfs/weed/filesys/file.go

361 lines
8.5 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"
"sort"
"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-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"
2020-03-23 07:01:34 +00:00
"github.com/chrislusf/seaweedfs/weed/util"
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.NodeIdentifier(&File{})
var _ = fs.NodeOpener(&File{})
var _ = fs.NodeFsyncer(&File{})
var _ = fs.NodeSetattrer(&File{})
2019-12-16 05:07:01 +00:00
var _ = fs.NodeGetxattrer(&File{})
var _ = fs.NodeSetxattrer(&File{})
var _ = fs.NodeRemovexattrer(&File{})
var _ = fs.NodeListxattrer(&File{})
var _ = fs.NodeForgetter(&File{})
2018-05-08 08:59:43 +00:00
2018-05-06 05:47:16 +00:00
type File struct {
2021-04-14 07:30:16 +00:00
Name string
dir *Dir
wfs *WFS
entry *filer_pb.Entry
isOpen int
dirtyMetadata bool
2021-04-17 17:48:22 +00:00
id uint64
2018-05-06 05:47:16 +00:00
}
2020-03-23 07:01:34 +00:00
func (file *File) fullpath() util.FullPath {
return util.NewFullPath(file.dir.FullPath(), file.Name)
2018-06-06 09:09:57 +00:00
}
func (file *File) Id() uint64 {
2021-04-17 17:48:22 +00:00
return file.id
}
func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
2021-04-17 17:48:22 +00:00
entry, err := file.maybeLoadEntry(ctx)
if err != nil {
return err
}
2018-05-08 08:59:43 +00:00
2021-03-16 09:55:49 +00:00
if entry == nil {
return fuse.ENOENT
}
2021-04-17 17:48:22 +00:00
attr.Inode = file.Id()
2020-01-20 07:59:46 +00:00
attr.Valid = time.Second
attr.Mode = os.FileMode(entry.Attributes.FileMode)
attr.Size = filer.FileSize(entry)
if file.isOpen > 0 {
attr.Size = entry.Attributes.FileSize
2020-01-25 08:32:18 +00:00
glog.V(4).Infof("file Attr %s, open:%v, size: %d", file.fullpath(), file.isOpen, attr.Size)
2020-01-19 20:07:04 +00:00
}
attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
attr.Gid = entry.Attributes.Gid
attr.Uid = entry.Attributes.Uid
attr.Blocks = attr.Size/blockSize + 1
2018-10-14 07:18:52 +00:00
attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
if entry.HardLinkCounter > 0 {
attr.Nlink = uint32(entry.HardLinkCounter)
2020-09-24 10:06:44 +00:00
}
2018-05-19 20:51:44 +00:00
return nil
2018-05-08 08:59:43 +00:00
}
2019-12-16 05:07:01 +00:00
func (file *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
2020-03-22 23:19:00 +00:00
glog.V(4).Infof("file Getxattr %s", file.fullpath())
2019-12-16 05:07:01 +00:00
entry, err := file.maybeLoadEntry(ctx)
if err != nil {
2019-12-16 05:07:01 +00:00
return err
}
return getxattr(entry, req, resp)
2019-12-16 05:07:01 +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
2020-08-19 06:42:09 +00:00
glog.V(4).Infof("file %v open %+v", file.fullpath(), req)
handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid, req.Flags&fuse.OpenWriteOnly > 0)
2018-06-06 09:09:57 +00:00
resp.Handle = fuse.HandleID(handle.handle)
2020-08-19 06:42:26 +00:00
glog.V(4).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
2018-06-06 09:09:57 +00:00
return handle, nil
}
func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
2020-08-31 03:12:04 +00:00
glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
2021-03-02 21:33:56 +00:00
entry, err := file.maybeLoadEntry(ctx)
if err != nil {
return err
}
if file.isOpen > 0 {
file.wfs.handlesLock.Lock()
2021-04-17 17:48:22 +00:00
fileHandle := file.wfs.handles[file.Id()]
file.wfs.handlesLock.Unlock()
if fileHandle != nil {
fileHandle.Lock()
defer fileHandle.Unlock()
}
}
if req.Valid.Size() {
2021-03-02 21:33:56 +00:00
glog.V(4).Infof("%v file setattr set size=%v chunks=%d", file.fullpath(), req.Size, len(entry.Chunks))
if req.Size < filer.FileSize(entry) {
2018-05-25 06:20:12 +00:00
// fmt.Printf("truncate %v \n", fullPath)
2020-01-20 08:00:08 +00:00
var chunks []*filer_pb.FileChunk
var truncatedChunks []*filer_pb.FileChunk
2021-03-02 21:33:56 +00:00
for _, chunk := range entry.Chunks {
int64Size := int64(chunk.Size)
if chunk.Offset+int64Size > int64(req.Size) {
// this chunk is truncated
int64Size = int64(req.Size) - chunk.Offset
if int64Size > 0 {
chunks = append(chunks, chunk)
2020-08-16 02:55:28 +00:00
glog.V(4).Infof("truncated chunk %+v from %d to %d\n", chunk.GetFileIdString(), chunk.Size, int64Size)
chunk.Size = uint64(int64Size)
} else {
2020-08-16 02:55:28 +00:00
glog.V(4).Infof("truncated whole chunk %+v\n", chunk.GetFileIdString())
truncatedChunks = append(truncatedChunks, chunk)
}
2020-01-20 08:00:08 +00:00
}
}
2021-03-02 21:33:56 +00:00
entry.Chunks = chunks
}
entry.Attributes.FileSize = req.Size
2020-08-16 04:09:31 +00:00
file.dirtyMetadata = true
}
2020-08-16 04:09:31 +00:00
if req.Valid.Mode() {
entry.Attributes.FileMode = uint32(req.Mode)
2020-08-15 21:15:07 +00:00
file.dirtyMetadata = true
}
if req.Valid.Uid() {
entry.Attributes.Uid = req.Uid
2020-08-15 21:15:07 +00:00
file.dirtyMetadata = true
}
if req.Valid.Gid() {
entry.Attributes.Gid = req.Gid
2020-08-15 21:15:07 +00:00
file.dirtyMetadata = true
}
2018-12-29 06:37:18 +00:00
if req.Valid.Crtime() {
entry.Attributes.Crtime = req.Crtime.Unix()
2020-08-15 21:15:07 +00:00
file.dirtyMetadata = true
2018-12-29 06:37:18 +00:00
}
if req.Valid.Mtime() {
entry.Attributes.Mtime = req.Mtime.Unix()
2020-08-15 21:15:07 +00:00
file.dirtyMetadata = true
}
if req.Valid.Handle() {
// fmt.Printf("file handle => %d\n", req.Handle)
}
if file.isOpen > 0 {
2019-02-15 18:00:27 +00:00
return nil
}
2020-08-16 02:56:24 +00:00
if !file.dirtyMetadata {
return nil
}
2021-03-02 21:33:56 +00:00
return file.saveEntry(entry)
2019-12-16 05:07:01 +00:00
}
2019-12-16 05:07:01 +00:00
func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
2019-12-16 05:07:01 +00:00
glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
entry, err := file.maybeLoadEntry(ctx)
if err != nil {
2019-12-16 05:07:01 +00:00
return err
}
if err := setxattr(entry, req); err != nil {
2019-12-16 05:07:01 +00:00
return err
}
return file.saveEntry(entry)
2018-05-19 20:51:44 +00:00
2018-05-06 05:47:16 +00:00
}
2019-12-16 05:07:01 +00:00
func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
2019-12-16 05:07:01 +00:00
glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
entry, err := file.maybeLoadEntry(ctx)
if err != nil {
2019-12-16 05:07:01 +00:00
return err
}
if err := removexattr(entry, req); err != nil {
2019-12-16 05:07:01 +00:00
return err
}
return file.saveEntry(entry)
2019-12-16 05:07:01 +00:00
}
2019-12-16 05:07:01 +00:00
func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp *fuse.ListxattrResponse) error {
2019-12-16 06:24:06 +00:00
glog.V(4).Infof("file Listxattr %s", file.fullpath())
entry, err := file.maybeLoadEntry(ctx)
if err != nil {
2019-12-16 05:07:01 +00:00
return err
}
if err := listxattr(entry, req, resp); err != nil {
2019-12-16 05:07:01 +00:00
return err
}
2019-12-18 04:38:56 +00:00
return nil
2018-11-07 19:35:13 +00:00
2019-12-16 05:07:01 +00:00
}
2019-12-16 05:07:01 +00:00
func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
// fsync works at OS level
// write the file chunks to the filerGrpcAddress
2020-08-14 07:22:21 +00:00
glog.V(4).Infof("%s/%s fsync file %+v", file.dir.FullPath(), file.Name, req)
2019-12-16 05:07:01 +00:00
return nil
}
func (file *File) Forget() {
2020-05-02 08:33:49 +00:00
t := util.NewFullPath(file.dir.FullPath(), file.Name)
2020-08-31 03:12:04 +00:00
glog.V(4).Infof("Forget file %s", t)
2021-04-21 02:56:51 +00:00
file.wfs.ReleaseHandle(t, fuse.HandleID(t.AsInode()))
}
func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
2021-04-17 17:48:22 +00:00
file.wfs.handlesLock.Lock()
handle, found := file.wfs.handles[file.Id()]
file.wfs.handlesLock.Unlock()
entry = file.entry
2021-04-17 17:48:22 +00:00
if found {
2021-05-06 10:34:34 +00:00
glog.V(4).Infof("maybeLoadEntry found opened file %s/%s", file.dir.FullPath(), file.Name)
2021-04-17 17:48:22 +00:00
entry = handle.f.entry
2020-10-25 17:33:49 +00:00
}
2021-04-17 17:48:22 +00:00
if entry != nil {
if len(entry.HardLinkId) == 0 {
// only always reload hard link
return entry, nil
}
}
entry, err = file.wfs.maybeLoadEntry(file.dir.FullPath(), file.Name)
2020-10-25 17:33:49 +00:00
if err != nil {
glog.V(3).Infof("maybeLoadEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
return entry, err
2020-10-25 17:33:49 +00:00
}
if entry != nil {
// file.entry = entry
} else {
glog.Warningf("maybeLoadEntry not found entry %s/%s: %v", file.dir.FullPath(), file.Name, err)
}
return entry, nil
}
2019-01-05 23:16:39 +00:00
func lessThan(a, b *filer_pb.FileChunk) bool {
if a.Mtime == b.Mtime {
return a.Fid.FileKey < b.Fid.FileKey
}
return a.Mtime < b.Mtime
}
2019-01-05 23:16:39 +00:00
func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
// find the earliest incoming chunk
newChunks := chunks
earliestChunk := newChunks[0]
2020-11-16 00:59:28 +00:00
for i := 1; i < len(newChunks); i++ {
if lessThan(earliestChunk, newChunks[i]) {
earliestChunk = newChunks[i]
}
}
entry := file.getEntry()
if entry == nil {
return
}
// pick out-of-order chunks from existing chunks
for _, chunk := range entry.Chunks {
if lessThan(earliestChunk, chunk) {
chunks = append(chunks, chunk)
}
}
// sort incoming chunks
sort.Slice(chunks, func(i, j int) bool {
return lessThan(chunks[i], chunks[j])
})
glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
2019-06-21 19:05:00 +00:00
entry.Chunks = append(entry.Chunks, newChunks...)
2019-01-05 23:16:39 +00:00
}
2019-01-05 23:21:56 +00:00
func (file *File) saveEntry(entry *filer_pb.Entry) error {
return file.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
2019-12-16 05:07:01 +00:00
file.wfs.mapPbIdFromLocalToFiler(entry)
defer file.wfs.mapPbIdFromFilerToLocal(entry)
request := &filer_pb.CreateEntryRequest{
Directory: file.dir.FullPath(),
Entry: entry,
Signatures: []int32{file.wfs.signature},
2019-12-16 05:07:01 +00:00
}
2020-08-16 04:09:31 +00:00
glog.V(4).Infof("save file entry: %v", request)
_, err := client.CreateEntry(context.Background(), request)
2019-12-16 05:07:01 +00:00
if err != nil {
2020-08-19 06:42:09 +00:00
glog.Errorf("UpdateEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
2019-12-16 05:07:01 +00:00
return fuse.EIO
}
file.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
2019-12-16 05:07:01 +00:00
return nil
})
}
func (file *File) getEntry() *filer_pb.Entry {
return file.entry
}