aboutsummaryrefslogtreecommitdiff
path: root/weed/filesys
diff options
context:
space:
mode:
authorChris Lu <chris.lu@gmail.com>2018-05-23 03:08:46 -0700
committerChris Lu <chris.lu@gmail.com>2018-05-23 03:08:46 -0700
commit536559f62dd937da7780e743cd7056e9682e2fb6 (patch)
tree3df89147a278880dafc597144919070430d18b14 /weed/filesys
parent1675243f29c5f37ca698b36b8bd4e74d5f1671f4 (diff)
downloadseaweedfs-536559f62dd937da7780e743cd7056e9682e2fb6.tar.xz
seaweedfs-536559f62dd937da7780e743cd7056e9682e2fb6.zip
copy works, edit somehow still fails
Diffstat (limited to 'weed/filesys')
-rw-r--r--weed/filesys/dir.go16
-rw-r--r--weed/filesys/file.go206
-rw-r--r--weed/filesys/filehandle.go176
3 files changed, 230 insertions, 168 deletions
diff --git a/weed/filesys/dir.go b/weed/filesys/dir.go
index 51b617689..8d07705c6 100644
--- a/weed/filesys/dir.go
+++ b/weed/filesys/dir.go
@@ -114,9 +114,19 @@ func (dir *Dir) Create(ctx context.Context, req *fuse.CreateRequest,
})
if err == nil {
- node := dir.newFile(req.Name, nil)
- dir.NodeMap[req.Name] = node
- return node, node, nil
+ file := dir.newFile(req.Name, nil)
+ dir.NodeMap[req.Name] = file
+ return file, &FileHandle{
+ wfs: file.wfs,
+ dirPath: file.dir.Path,
+ name: file.Name,
+ RequestId: req.Header.ID,
+ NodeId: req.Header.Node,
+ Uid: req.Uid,
+ Gid: req.Gid,
+ attributes: file.attributes,
+ Chunks: file.Chunks,
+ }, nil
}
return nil, nil, err
diff --git a/weed/filesys/file.go b/weed/filesys/file.go
index 9604d6fb0..55edaf515 100644
--- a/weed/filesys/file.go
+++ b/weed/filesys/file.go
@@ -11,20 +11,12 @@ import (
"path/filepath"
"os"
"time"
- "bytes"
- "github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/filer2"
)
var _ = fs.Node(&File{})
var _ = fs.NodeOpener(&File{})
var _ = fs.NodeFsyncer(&File{})
-var _ = fs.Handle(&File{})
-var _ = fs.HandleReadAller(&File{})
-// var _ = fs.HandleReader(&File{})
-var _ = fs.HandleFlusher(&File{})
-var _ = fs.HandleWriter(&File{})
-var _ = fs.HandleReleaser(&File{})
var _ = fs.NodeSetattrer(&File{})
type File struct {
@@ -32,44 +24,42 @@ type File struct {
Name string
dir *Dir
wfs *WFS
- isOpened bool
attributes *filer_pb.FuseAttributes
}
func (file *File) Attr(context context.Context, attr *fuse.Attr) error {
- if !file.isOpened {
- fullPath := filepath.Join(file.dir.Path, file.Name)
- item := file.wfs.listDirectoryEntriesCache.Get(fullPath)
- if item != nil {
- entry := item.Value().(*filer_pb.Entry)
- file.Chunks = entry.Chunks
- file.attributes = entry.Attributes
- glog.V(1).Infof("read cached file %v attributes", file.Name)
- } else {
- err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+ fullPath := filepath.Join(file.dir.Path, file.Name)
+ item := file.wfs.listDirectoryEntriesCache.Get(fullPath)
+ if item != nil {
+ entry := item.Value().(*filer_pb.Entry)
+ file.Chunks = entry.Chunks
+ file.attributes = entry.Attributes
+ 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.GetEntryAttributesRequest{
+ Name: file.Name,
+ ParentDir: file.dir.Path,
+ }
- request := &filer_pb.GetEntryAttributesRequest{
- Name: file.Name,
- ParentDir: file.dir.Path,
- }
+ resp, err := client.GetEntryAttributes(context, request)
+ if err != nil {
+ glog.V(0).Infof("file attr read file %v: %v", request, err)
+ return err
+ }
- glog.V(1).Infof("read file: %v", request)
- resp, err := client.GetEntryAttributes(context, request)
- if err != nil {
- glog.V(0).Infof("read file %v: %v", request, err)
- return err
- }
+ file.attributes = resp.Attributes
+ file.Chunks = resp.Chunks
- file.attributes = resp.Attributes
- file.Chunks = resp.Chunks
+ glog.V(1).Infof("file attr %v %+v: %d", fullPath, file.attributes, filer2.TotalSize(file.Chunks))
- return nil
- })
+ return nil
+ })
- if err != nil {
- return err
- }
+ if err != nil {
+ return err
}
}
@@ -84,19 +74,29 @@ func (file *File) Attr(context context.Context, attr *fuse.Attr) error {
}
func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
+
fullPath := filepath.Join(file.dir.Path, file.Name)
- fmt.Printf("Open %v %+v\n", fullPath, req)
- file.isOpened = true
+ glog.V(3).Infof("file open %v %+v", fullPath, req)
- return file, nil
+ return &FileHandle{
+ wfs: file.wfs,
+ dirPath: file.dir.Path,
+ name: file.Name,
+ RequestId: req.Header.ID,
+ NodeId: req.Header.Node,
+ Uid: req.Uid,
+ Gid: req.Gid,
+ attributes: file.attributes,
+ Chunks: file.Chunks,
+ }, nil
}
func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
fullPath := filepath.Join(file.dir.Path, file.Name)
- fmt.Printf("Setattr %v %+v\n", fullPath, req)
+ glog.V(3).Infof("file setattr %v %+v", fullPath, req)
if req.Valid.Size() {
if req.Size == 0 {
@@ -125,134 +125,10 @@ func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *f
}
-func (file *File) ReadAll(ctx context.Context) (content []byte, err error) {
-
- fmt.Printf("read all file %+v/%v\n", file.dir.Path, file.Name)
-
- if len(file.Chunks) == 0 {
- glog.V(0).Infof("empty file %v/%v", file.dir.Path, file.Name)
- return
- }
-
- err = file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
- // FIXME: need to either use Read() or implement differently
- chunks, _ := filer2.CompactFileChunks(file.Chunks)
- glog.V(1).Infof("read file %v/%v %d/%d chunks", file.dir.Path, file.Name, len(chunks), len(file.Chunks))
- request := &filer_pb.GetFileContentRequest{
- FileId: chunks[0].FileId,
- }
-
- glog.V(1).Infof("read file content %d chunk %s [%d,%d): %v", len(chunks),
- chunks[0].FileId, chunks[0].Offset, chunks[0].Offset+int64(chunks[0].Size), request)
- resp, err := client.GetFileContent(ctx, request)
- if err != nil {
- return err
- }
-
- content = resp.Content
-
- return nil
- })
-
- return content, err
-}
-
func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
// fsync works at OS level
// write the file chunks to the filer
- fmt.Printf("flush file %+v\n", req)
-
- return nil
-}
-
-func (file *File) Flush(ctx context.Context, req *fuse.FlushRequest) error {
- // fflush works at file level
- // send the data to the OS
- glog.V(3).Infof("file flush %v", req)
-
- if len(file.Chunks) == 0 {
- glog.V(2).Infof("%x file %s/%s flush skipping empty: %v", file, file.dir.Path, file.Name, req)
- return nil
- }
-
- err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
- request := &filer_pb.UpdateEntryRequest{
- Directory: file.dir.Path,
- Entry: &filer_pb.Entry{
- Name: file.Name,
- Attributes: file.attributes,
- Chunks: file.Chunks,
- },
- }
-
- glog.V(1).Infof("%s/%s set chunks: %v", file.dir.Path, file.Name, len(file.Chunks))
- if _, err := client.UpdateEntry(ctx, request); err != nil {
- return fmt.Errorf("update file: %v", err)
- }
-
- return nil
- })
-
- return err
-}
-
-func (file *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
- // write the request to volume servers
- // fmt.Printf("write file %+v\n", req)
-
- var fileId, host string
-
- if err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
-
- request := &filer_pb.AssignVolumeRequest{
- Count: 1,
- Replication: "000",
- Collection: "",
- }
-
- glog.V(1).Infof("assign volume: %v", request)
- resp, err := client.AssignVolume(ctx, request)
- if err != nil {
- return err
- }
-
- fileId, host = resp.FileId, resp.Url
-
- return nil
- }); err != nil {
- return fmt.Errorf("filer assign volume: %v", err)
- }
-
- fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
- bufReader := bytes.NewReader(req.Data)
- uploadResult, err := operation.Upload(fileUrl, file.Name, bufReader, false, "application/octet-stream", nil, "")
- if err != nil {
- return fmt.Errorf("upload data: %v", err)
- }
- if uploadResult.Error != "" {
- return fmt.Errorf("upload result: %v", uploadResult.Error)
- }
-
- resp.Size = int(uploadResult.Size)
-
- file.Chunks = append(file.Chunks, &filer_pb.FileChunk{
- FileId: fileId,
- Offset: req.Offset,
- Size: uint64(uploadResult.Size),
- Mtime: time.Now().UnixNano(),
- })
-
- glog.V(1).Infof("uploaded %s/%s to: %v, [%d,%d)", file.dir.Path, file.Name, fileUrl, req.Offset, req.Offset+int64(resp.Size))
-
- return nil
-}
-
-func (file *File) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
-
- fmt.Printf("release file %+v\n", req)
- file.isOpened = false
+ glog.V(3).Infof("fsync file %+v\n", req)
return nil
}
diff --git a/weed/filesys/filehandle.go b/weed/filesys/filehandle.go
new file mode 100644
index 000000000..5599f53a8
--- /dev/null
+++ b/weed/filesys/filehandle.go
@@ -0,0 +1,176 @@
+package filesys
+
+import (
+ "bazil.org/fuse/fs"
+ "fmt"
+ "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+ "github.com/chrislusf/seaweedfs/weed/filer2"
+ "context"
+ "github.com/chrislusf/seaweedfs/weed/glog"
+ "bazil.org/fuse"
+ "bytes"
+ "github.com/chrislusf/seaweedfs/weed/operation"
+ "time"
+)
+
+type FileHandle struct {
+ // cache file has been written to
+ dirty bool
+
+ cachePath string
+
+ handle uint64
+
+ wfs *WFS
+ dirPath string
+ name string
+ 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
+ attributes *filer_pb.FuseAttributes
+ Chunks []*filer_pb.FileChunk
+}
+
+var _ = fs.Handle(&FileHandle{})
+var _ = fs.HandleReadAller(&FileHandle{})
+// var _ = fs.HandleReader(&FileHandle{})
+var _ = fs.HandleFlusher(&FileHandle{})
+var _ = fs.HandleWriter(&FileHandle{})
+var _ = fs.HandleReleaser(&FileHandle{})
+
+func (fh *FileHandle) ReadAll(ctx context.Context) (content []byte, err error) {
+
+ glog.V(3).Infof("read all fh %+v/%v", fh.dirPath, fh.name)
+
+ if len(fh.Chunks) == 0 {
+ glog.V(0).Infof("empty fh %v/%v", fh.dirPath, fh.name)
+ return
+ }
+
+ err = fh.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+ // FIXME: need to either use Read() or implement differently
+ chunks, _ := filer2.CompactFileChunks(fh.Chunks)
+ glog.V(1).Infof("read fh %v/%v %d/%d chunks", fh.dirPath, fh.name, len(chunks), len(fh.Chunks))
+ request := &filer_pb.GetFileContentRequest{
+ FileId: chunks[0].FileId,
+ }
+
+ glog.V(1).Infof("read fh content %d chunk %s [%d,%d): %v", len(chunks),
+ chunks[0].FileId, chunks[0].Offset, chunks[0].Offset+int64(chunks[0].Size), request)
+ resp, err := client.GetFileContent(ctx, request)
+ if err != nil {
+ return err
+ }
+
+ content = resp.Content
+
+ return nil
+ })
+
+ return content, err
+}
+
+// Write to the file handle
+func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
+ // write the request to volume servers
+ // glog.V(3).Infof("write fh %+v", req)
+
+ var fileId, host string
+
+ if err := fh.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+ request := &filer_pb.AssignVolumeRequest{
+ Count: 1,
+ Replication: "000",
+ Collection: "",
+ }
+
+ glog.V(1).Infof("assign volume: %v", request)
+ resp, err := client.AssignVolume(ctx, request)
+ if err != nil {
+ return err
+ }
+
+ fileId, host = resp.FileId, resp.Url
+
+ return nil
+ }); err != nil {
+ return fmt.Errorf("filer assign volume: %v", err)
+ }
+
+ fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
+ bufReader := bytes.NewReader(req.Data)
+ uploadResult, err := operation.Upload(fileUrl, fh.name, bufReader, false, "application/octet-stream", nil, "")
+ if err != nil {
+ return fmt.Errorf("upload data: %v", err)
+ }
+ if uploadResult.Error != "" {
+ return fmt.Errorf("upload result: %v", uploadResult.Error)
+ }
+
+ resp.Size = int(uploadResult.Size)
+
+ fh.Chunks = append(fh.Chunks, &filer_pb.FileChunk{
+ FileId: fileId,
+ Offset: req.Offset,
+ Size: uint64(uploadResult.Size),
+ Mtime: time.Now().UnixNano(),
+ })
+
+ glog.V(1).Infof("uploaded %s/%s to: %v, [%d,%d)", fh.dirPath, fh.name, fileUrl, req.Offset, req.Offset+int64(resp.Size))
+
+ fh.dirty = true
+
+ return nil
+}
+
+func (fh *FileHandle) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
+
+ glog.V(3).Infof("release fh %+v/%v", fh.dirPath, fh.name)
+
+ return nil
+}
+
+// Flush - experimenting with uploading at flush, this slows operations down till it has been
+// completely flushed
+func (fh *FileHandle) Flush(ctx context.Context, req *fuse.FlushRequest) error {
+ // fflush works at fh level
+ // send the data to the OS
+ glog.V(3).Infof("fh flush %v", req)
+
+ if !fh.dirty {
+ return nil
+ }
+
+ if len(fh.Chunks) == 0 {
+ glog.V(2).Infof("fh %s/%s flush skipping empty: %v", fh.dirPath, fh.name, req)
+ return nil
+ }
+
+ err := fh.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+
+ request := &filer_pb.UpdateEntryRequest{
+ Directory: fh.dirPath,
+ Entry: &filer_pb.Entry{
+ Name: fh.name,
+ Attributes: fh.attributes,
+ Chunks: fh.Chunks,
+ },
+ }
+
+ glog.V(1).Infof("%s/%s set chunks: %v", fh.dirPath, fh.name, len(fh.Chunks))
+ if _, err := client.UpdateEntry(ctx, request); err != nil {
+ return fmt.Errorf("update fh: %v", err)
+ }
+
+ return nil
+ })
+
+ if err == nil {
+ fh.dirty = false
+ }
+
+ return err
+}