You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

216 lines
5.3 KiB

7 years ago
7 years ago
7 years ago
6 years ago
5 years ago
5 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "os"
  5. "path/filepath"
  6. "sort"
  7. "time"
  8. "github.com/chrislusf/seaweedfs/weed/filer2"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  11. "github.com/seaweedfs/fuse"
  12. "github.com/seaweedfs/fuse/fs"
  13. )
  14. const blockSize = 512
  15. var _ = fs.Node(&File{})
  16. var _ = fs.NodeOpener(&File{})
  17. var _ = fs.NodeFsyncer(&File{})
  18. var _ = fs.NodeSetattrer(&File{})
  19. type File struct {
  20. Name string
  21. dir *Dir
  22. wfs *WFS
  23. entry *filer_pb.Entry
  24. entryViewCache []filer2.VisibleInterval
  25. isOpen bool
  26. }
  27. func (file *File) fullpath() string {
  28. return filepath.Join(file.dir.Path, file.Name)
  29. }
  30. func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error {
  31. glog.V(4).Infof("file Attr %s", file.fullpath())
  32. if err := file.maybeLoadAttributes(ctx); err != nil {
  33. return err
  34. }
  35. attr.Mode = os.FileMode(file.entry.Attributes.FileMode)
  36. attr.Size = filer2.TotalSize(file.entry.Chunks)
  37. attr.Mtime = time.Unix(file.entry.Attributes.Mtime, 0)
  38. attr.Gid = file.entry.Attributes.Gid
  39. attr.Uid = file.entry.Attributes.Uid
  40. attr.Blocks = attr.Size/blockSize + 1
  41. attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
  42. return nil
  43. }
  44. func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
  45. glog.V(4).Infof("file %v open %+v", file.fullpath(), req)
  46. file.isOpen = true
  47. handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid)
  48. resp.Handle = fuse.HandleID(handle.handle)
  49. glog.V(3).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
  50. return handle, nil
  51. }
  52. func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
  53. if err := file.maybeLoadAttributes(ctx); err != nil {
  54. return err
  55. }
  56. glog.V(3).Infof("%v file setattr %+v, old:%+v", file.fullpath(), req, file.entry.Attributes)
  57. if req.Valid.Size() {
  58. glog.V(3).Infof("%v file setattr set size=%v", file.fullpath(), req.Size)
  59. if req.Size == 0 {
  60. // fmt.Printf("truncate %v \n", fullPath)
  61. file.entry.Chunks = nil
  62. file.entryViewCache = nil
  63. }
  64. file.entry.Attributes.FileSize = req.Size
  65. }
  66. if req.Valid.Mode() {
  67. file.entry.Attributes.FileMode = uint32(req.Mode)
  68. }
  69. if req.Valid.Uid() {
  70. file.entry.Attributes.Uid = req.Uid
  71. }
  72. if req.Valid.Gid() {
  73. file.entry.Attributes.Gid = req.Gid
  74. }
  75. if req.Valid.Crtime() {
  76. file.entry.Attributes.Crtime = req.Crtime.Unix()
  77. }
  78. if req.Valid.Mtime() {
  79. file.entry.Attributes.Mtime = req.Mtime.Unix()
  80. }
  81. if file.isOpen {
  82. return nil
  83. }
  84. return file.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  85. request := &filer_pb.UpdateEntryRequest{
  86. Directory: file.dir.Path,
  87. Entry: file.entry,
  88. }
  89. glog.V(1).Infof("set attr file entry: %v", request)
  90. _, err := client.UpdateEntry(ctx, request)
  91. if err != nil {
  92. glog.V(0).Infof("UpdateEntry file %s/%s: %v", file.dir.Path, file.Name, err)
  93. return fuse.EIO
  94. }
  95. return nil
  96. })
  97. }
  98. func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
  99. // fsync works at OS level
  100. // write the file chunks to the filerGrpcAddress
  101. glog.V(3).Infof("%s/%s fsync file %+v", file.dir.Path, file.Name, req)
  102. return nil
  103. }
  104. func (file *File) maybeLoadAttributes(ctx context.Context) error {
  105. if file.entry == nil || !file.isOpen {
  106. item := file.wfs.listDirectoryEntriesCache.Get(file.fullpath())
  107. var entry *filer_pb.Entry
  108. if item != nil && !item.Expired() {
  109. entry = item.Value().(*filer_pb.Entry)
  110. }
  111. if entry != nil {
  112. glog.V(4).Infof("file read attr cache hit %s", file.fullpath())
  113. file.setEntry(entry)
  114. // glog.V(1).Infof("file attr read cached %v attributes", file.Name)
  115. } else {
  116. glog.V(3).Infof("file read attr cache miss %s", file.fullpath())
  117. err := file.wfs.WithFilerClient(ctx, func(client filer_pb.SeaweedFilerClient) error {
  118. request := &filer_pb.LookupDirectoryEntryRequest{
  119. Name: file.Name,
  120. Directory: file.dir.Path,
  121. }
  122. resp, err := client.LookupDirectoryEntry(ctx, request)
  123. if err != nil {
  124. glog.V(3).Infof("file attr read file %v: %v", request, err)
  125. return fuse.ENOENT
  126. }
  127. file.setEntry(resp.Entry)
  128. glog.V(3).Infof("file attr %v %+v: %d", file.fullpath(), file.entry.Attributes, filer2.TotalSize(file.entry.Chunks))
  129. // file.wfs.listDirectoryEntriesCache.Set(file.fullpath(), file.entry, file.wfs.option.EntryCacheTtl)
  130. return nil
  131. })
  132. if err != nil {
  133. return err
  134. }
  135. }
  136. }
  137. return nil
  138. }
  139. func (file *File) addChunk(chunk *filer_pb.FileChunk) {
  140. if chunk != nil {
  141. file.addChunks([]*filer_pb.FileChunk{chunk})
  142. }
  143. }
  144. func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
  145. sort.Slice(chunks, func(i, j int) bool {
  146. return chunks[i].Mtime < chunks[j].Mtime
  147. })
  148. var newVisibles []filer2.VisibleInterval
  149. for _, chunk := range chunks {
  150. newVisibles = filer2.MergeIntoVisibles(file.entryViewCache, newVisibles, chunk)
  151. t := file.entryViewCache[:0]
  152. file.entryViewCache = newVisibles
  153. newVisibles = t
  154. }
  155. glog.V(3).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
  156. file.entry.Chunks = append(file.entry.Chunks, chunks...)
  157. }
  158. func (file *File) setEntry(entry *filer_pb.Entry) {
  159. file.entry = entry
  160. file.entryViewCache = filer2.NonOverlappingVisibleIntervals(file.entry.Chunks)
  161. }