file.go 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276
  1. package filesys
  2. import (
  3. "context"
  4. "os"
  5. "sort"
  6. "time"
  7. "github.com/chrislusf/seaweedfs/weed/filer2"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/seaweedfs/fuse"
  11. "github.com/seaweedfs/fuse/fs"
  12. )
  13. const blockSize = 512
  14. var _ = fs.Node(&File{})
  15. var _ = fs.NodeOpener(&File{})
  16. var _ = fs.NodeFsyncer(&File{})
  17. var _ = fs.NodeSetattrer(&File{})
  18. var _ = fs.NodeGetxattrer(&File{})
  19. var _ = fs.NodeSetxattrer(&File{})
  20. var _ = fs.NodeRemovexattrer(&File{})
  21. var _ = fs.NodeListxattrer(&File{})
  22. var _ = fs.NodeForgetter(&File{})
  23. type File struct {
  24. Name string
  25. dir *Dir
  26. wfs *WFS
  27. entry *filer_pb.Entry
  28. entryViewCache []filer2.VisibleInterval
  29. isOpen int
  30. }
  31. func (file *File) fullpath() filer2.FullPath {
  32. return filer2.NewFullPath(file.dir.Path, file.Name)
  33. }
  34. func (file *File) Attr(ctx context.Context, attr *fuse.Attr) error {
  35. glog.V(4).Infof("file Attr %s, open:%v, existing attr: %+v", file.fullpath(), file.isOpen, attr)
  36. if file.isOpen <= 0 {
  37. if err := file.maybeLoadEntry(ctx); err != nil {
  38. return err
  39. }
  40. }
  41. attr.Inode = file.fullpath().AsInode()
  42. attr.Valid = time.Second
  43. attr.Mode = os.FileMode(file.entry.Attributes.FileMode)
  44. attr.Size = filer2.TotalSize(file.entry.Chunks)
  45. if file.isOpen > 0 {
  46. attr.Size = file.entry.Attributes.FileSize
  47. glog.V(4).Infof("file Attr %s, open:%v, size: %d", file.fullpath(), file.isOpen, attr.Size)
  48. }
  49. attr.Crtime = time.Unix(file.entry.Attributes.Crtime, 0)
  50. attr.Mtime = time.Unix(file.entry.Attributes.Mtime, 0)
  51. attr.Gid = file.entry.Attributes.Gid
  52. attr.Uid = file.entry.Attributes.Uid
  53. attr.Blocks = attr.Size/blockSize + 1
  54. attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
  55. return nil
  56. }
  57. func (file *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
  58. // glog.V(4).Infof("file Getxattr %s", file.fullpath())
  59. if err := file.maybeLoadEntry(ctx); err != nil {
  60. return err
  61. }
  62. return getxattr(file.entry, req, resp)
  63. }
  64. func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
  65. glog.V(4).Infof("file %v open %+v", file.fullpath(), req)
  66. file.isOpen++
  67. handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid)
  68. resp.Handle = fuse.HandleID(handle.handle)
  69. glog.V(3).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
  70. return handle, nil
  71. }
  72. func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
  73. glog.V(3).Infof("%v file setattr %+v, old:%+v", file.fullpath(), req, file.entry.Attributes)
  74. if err := file.maybeLoadEntry(ctx); err != nil {
  75. return err
  76. }
  77. if req.Valid.Size() {
  78. glog.V(3).Infof("%v file setattr set size=%v", file.fullpath(), req.Size)
  79. if req.Size < filer2.TotalSize(file.entry.Chunks) {
  80. // fmt.Printf("truncate %v \n", fullPath)
  81. var chunks []*filer_pb.FileChunk
  82. for _, chunk := range file.entry.Chunks {
  83. int64Size := int64(chunk.Size)
  84. if chunk.Offset+int64Size > int64(req.Size) {
  85. int64Size = int64(req.Size) - chunk.Offset
  86. }
  87. if int64Size > 0 {
  88. chunks = append(chunks, chunk)
  89. }
  90. }
  91. file.entry.Chunks = chunks
  92. file.entryViewCache = nil
  93. }
  94. file.entry.Attributes.FileSize = req.Size
  95. }
  96. if req.Valid.Mode() {
  97. file.entry.Attributes.FileMode = uint32(req.Mode)
  98. }
  99. if req.Valid.Uid() {
  100. file.entry.Attributes.Uid = req.Uid
  101. }
  102. if req.Valid.Gid() {
  103. file.entry.Attributes.Gid = req.Gid
  104. }
  105. if req.Valid.Crtime() {
  106. file.entry.Attributes.Crtime = req.Crtime.Unix()
  107. }
  108. if req.Valid.Mtime() {
  109. file.entry.Attributes.Mtime = req.Mtime.Unix()
  110. }
  111. if file.isOpen > 0 {
  112. return nil
  113. }
  114. file.wfs.cacheDelete(file.fullpath())
  115. return file.saveEntry(ctx)
  116. }
  117. func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
  118. glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
  119. if err := file.maybeLoadEntry(ctx); err != nil {
  120. return err
  121. }
  122. if err := setxattr(file.entry, req); err != nil {
  123. return err
  124. }
  125. file.wfs.cacheDelete(file.fullpath())
  126. return file.saveEntry(ctx)
  127. }
  128. func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
  129. glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
  130. if err := file.maybeLoadEntry(ctx); err != nil {
  131. return err
  132. }
  133. if err := removexattr(file.entry, req); err != nil {
  134. return err
  135. }
  136. file.wfs.cacheDelete(file.fullpath())
  137. return file.saveEntry(ctx)
  138. }
  139. func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp *fuse.ListxattrResponse) error {
  140. glog.V(4).Infof("file Listxattr %s", file.fullpath())
  141. if err := file.maybeLoadEntry(ctx); err != nil {
  142. return err
  143. }
  144. if err := listxattr(file.entry, req, resp); err != nil {
  145. return err
  146. }
  147. return nil
  148. }
  149. func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
  150. // fsync works at OS level
  151. // write the file chunks to the filerGrpcAddress
  152. glog.V(3).Infof("%s/%s fsync file %+v", file.dir.Path, file.Name, req)
  153. return nil
  154. }
  155. func (file *File) Forget() {
  156. glog.V(3).Infof("Forget file %s/%s", file.dir.Path, file.Name)
  157. file.wfs.forgetNode(filer2.NewFullPath(file.dir.Path, file.Name))
  158. }
  159. func (file *File) maybeLoadEntry(ctx context.Context) error {
  160. if file.entry == nil || file.isOpen <= 0 {
  161. entry, err := file.wfs.maybeLoadEntry(ctx, file.dir.Path, file.Name)
  162. if err != nil {
  163. return err
  164. }
  165. if entry != nil {
  166. file.setEntry(entry)
  167. }
  168. }
  169. return nil
  170. }
  171. func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
  172. sort.Slice(chunks, func(i, j int) bool {
  173. return chunks[i].Mtime < chunks[j].Mtime
  174. })
  175. var newVisibles []filer2.VisibleInterval
  176. for _, chunk := range chunks {
  177. newVisibles = filer2.MergeIntoVisibles(file.entryViewCache, newVisibles, chunk)
  178. t := file.entryViewCache[:0]
  179. file.entryViewCache = newVisibles
  180. newVisibles = t
  181. }
  182. glog.V(3).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
  183. file.entry.Chunks = append(file.entry.Chunks, chunks...)
  184. }
  185. func (file *File) setEntry(entry *filer_pb.Entry) {
  186. file.entry = entry
  187. file.entryViewCache = filer2.NonOverlappingVisibleIntervals(file.entry.Chunks)
  188. }
  189. func (file *File) saveEntry(ctx context.Context) error {
  190. return file.wfs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
  191. request := &filer_pb.UpdateEntryRequest{
  192. Directory: file.dir.Path,
  193. Entry: file.entry,
  194. }
  195. glog.V(1).Infof("save file entry: %v", request)
  196. _, err := client.UpdateEntry(ctx, request)
  197. if err != nil {
  198. glog.V(0).Infof("UpdateEntry file %s/%s: %v", file.dir.Path, file.Name, err)
  199. return fuse.EIO
  200. }
  201. return nil
  202. })
  203. }