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.

378 lines
9.1 KiB

7 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
4 years ago
4 years ago
7 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 years ago
6 years ago
6 years ago
4 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
6 years ago
4 years ago
4 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "io"
  5. "os"
  6. "sort"
  7. "sync"
  8. "time"
  9. "github.com/seaweedfs/fuse"
  10. "github.com/seaweedfs/fuse/fs"
  11. "github.com/chrislusf/seaweedfs/weed/filer"
  12. "github.com/chrislusf/seaweedfs/weed/glog"
  13. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  14. "github.com/chrislusf/seaweedfs/weed/util"
  15. )
  16. const blockSize = 512
  17. var _ = fs.Node(&File{})
  18. var _ = fs.NodeOpener(&File{})
  19. var _ = fs.NodeFsyncer(&File{})
  20. var _ = fs.NodeSetattrer(&File{})
  21. var _ = fs.NodeGetxattrer(&File{})
  22. var _ = fs.NodeSetxattrer(&File{})
  23. var _ = fs.NodeRemovexattrer(&File{})
  24. var _ = fs.NodeListxattrer(&File{})
  25. var _ = fs.NodeForgetter(&File{})
  26. type File struct {
  27. Name string
  28. dir *Dir
  29. wfs *WFS
  30. entry *filer_pb.Entry
  31. entryLock sync.RWMutex
  32. entryViewCache []filer.VisibleInterval
  33. isOpen int
  34. reader io.ReaderAt
  35. dirtyMetadata bool
  36. }
  37. func (file *File) fullpath() util.FullPath {
  38. return util.NewFullPath(file.dir.FullPath(), file.Name)
  39. }
  40. func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
  41. glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
  42. entry := file.getEntry()
  43. if file.isOpen <= 0 || entry == nil {
  44. if entry, err = file.maybeLoadEntry(ctx); err != nil {
  45. return err
  46. }
  47. }
  48. // attr.Inode = file.fullpath().AsInode()
  49. attr.Valid = time.Second
  50. attr.Mode = os.FileMode(entry.Attributes.FileMode)
  51. attr.Size = filer.FileSize(entry)
  52. if file.isOpen > 0 {
  53. attr.Size = entry.Attributes.FileSize
  54. glog.V(4).Infof("file Attr %s, open:%v, size: %d", file.fullpath(), file.isOpen, attr.Size)
  55. }
  56. attr.Crtime = time.Unix(entry.Attributes.Crtime, 0)
  57. attr.Mtime = time.Unix(entry.Attributes.Mtime, 0)
  58. attr.Gid = entry.Attributes.Gid
  59. attr.Uid = entry.Attributes.Uid
  60. attr.Blocks = attr.Size/blockSize + 1
  61. attr.BlockSize = uint32(file.wfs.option.ChunkSizeLimit)
  62. if entry.HardLinkCounter > 0 {
  63. attr.Nlink = uint32(entry.HardLinkCounter)
  64. }
  65. return nil
  66. }
  67. func (file *File) Getxattr(ctx context.Context, req *fuse.GetxattrRequest, resp *fuse.GetxattrResponse) error {
  68. glog.V(4).Infof("file Getxattr %s", file.fullpath())
  69. entry, err := file.maybeLoadEntry(ctx)
  70. if err != nil {
  71. return err
  72. }
  73. return getxattr(entry, req, resp)
  74. }
  75. func (file *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
  76. glog.V(4).Infof("file %v open %+v", file.fullpath(), req)
  77. handle := file.wfs.AcquireHandle(file, req.Uid, req.Gid)
  78. resp.Handle = fuse.HandleID(handle.handle)
  79. glog.V(4).Infof("%v file open handle id = %d", file.fullpath(), handle.handle)
  80. return handle, nil
  81. }
  82. func (file *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fuse.SetattrResponse) error {
  83. glog.V(4).Infof("%v file setattr %+v", file.fullpath(), req)
  84. entry, err := file.maybeLoadEntry(ctx)
  85. if err != nil {
  86. return err
  87. }
  88. if file.isOpen > 0 {
  89. file.wfs.handlesLock.Lock()
  90. fileHandle := file.wfs.handles[file.fullpath().AsInode()]
  91. file.wfs.handlesLock.Unlock()
  92. if fileHandle != nil {
  93. fileHandle.Lock()
  94. defer fileHandle.Unlock()
  95. }
  96. }
  97. if req.Valid.Size() {
  98. glog.V(4).Infof("%v file setattr set size=%v chunks=%d", file.fullpath(), req.Size, len(entry.Chunks))
  99. if req.Size < filer.FileSize(entry) {
  100. // fmt.Printf("truncate %v \n", fullPath)
  101. var chunks []*filer_pb.FileChunk
  102. var truncatedChunks []*filer_pb.FileChunk
  103. for _, chunk := range entry.Chunks {
  104. int64Size := int64(chunk.Size)
  105. if chunk.Offset+int64Size > int64(req.Size) {
  106. // this chunk is truncated
  107. int64Size = int64(req.Size) - chunk.Offset
  108. if int64Size > 0 {
  109. chunks = append(chunks, chunk)
  110. glog.V(4).Infof("truncated chunk %+v from %d to %d\n", chunk.GetFileIdString(), chunk.Size, int64Size)
  111. chunk.Size = uint64(int64Size)
  112. } else {
  113. glog.V(4).Infof("truncated whole chunk %+v\n", chunk.GetFileIdString())
  114. truncatedChunks = append(truncatedChunks, chunk)
  115. }
  116. }
  117. }
  118. entry.Chunks = chunks
  119. file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), chunks)
  120. file.reader = nil
  121. }
  122. entry.Attributes.FileSize = req.Size
  123. file.dirtyMetadata = true
  124. }
  125. if req.Valid.Mode() {
  126. entry.Attributes.FileMode = uint32(req.Mode)
  127. file.dirtyMetadata = true
  128. }
  129. if req.Valid.Uid() {
  130. entry.Attributes.Uid = req.Uid
  131. file.dirtyMetadata = true
  132. }
  133. if req.Valid.Gid() {
  134. entry.Attributes.Gid = req.Gid
  135. file.dirtyMetadata = true
  136. }
  137. if req.Valid.Crtime() {
  138. entry.Attributes.Crtime = req.Crtime.Unix()
  139. file.dirtyMetadata = true
  140. }
  141. if req.Valid.Mtime() {
  142. entry.Attributes.Mtime = req.Mtime.Unix()
  143. file.dirtyMetadata = true
  144. }
  145. if req.Valid.Handle() {
  146. // fmt.Printf("file handle => %d\n", req.Handle)
  147. }
  148. if file.isOpen > 0 {
  149. return nil
  150. }
  151. if !file.dirtyMetadata {
  152. return nil
  153. }
  154. return file.saveEntry(entry)
  155. }
  156. func (file *File) Setxattr(ctx context.Context, req *fuse.SetxattrRequest) error {
  157. glog.V(4).Infof("file Setxattr %s: %s", file.fullpath(), req.Name)
  158. entry, err := file.maybeLoadEntry(ctx)
  159. if err != nil {
  160. return err
  161. }
  162. if err := setxattr(entry, req); err != nil {
  163. return err
  164. }
  165. return file.saveEntry(entry)
  166. }
  167. func (file *File) Removexattr(ctx context.Context, req *fuse.RemovexattrRequest) error {
  168. glog.V(4).Infof("file Removexattr %s: %s", file.fullpath(), req.Name)
  169. entry, err := file.maybeLoadEntry(ctx)
  170. if err != nil {
  171. return err
  172. }
  173. if err := removexattr(entry, req); err != nil {
  174. return err
  175. }
  176. return file.saveEntry(entry)
  177. }
  178. func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, resp *fuse.ListxattrResponse) error {
  179. glog.V(4).Infof("file Listxattr %s", file.fullpath())
  180. entry, err := file.maybeLoadEntry(ctx)
  181. if err != nil {
  182. return err
  183. }
  184. if err := listxattr(entry, req, resp); err != nil {
  185. return err
  186. }
  187. return nil
  188. }
  189. func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
  190. // fsync works at OS level
  191. // write the file chunks to the filerGrpcAddress
  192. glog.V(4).Infof("%s/%s fsync file %+v", file.dir.FullPath(), file.Name, req)
  193. return nil
  194. }
  195. func (file *File) Forget() {
  196. t := util.NewFullPath(file.dir.FullPath(), file.Name)
  197. glog.V(4).Infof("Forget file %s", t)
  198. file.wfs.fsNodeCache.DeleteFsNode(t)
  199. }
  200. func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
  201. entry = file.getEntry()
  202. if file.isOpen > 0 {
  203. return entry, nil
  204. }
  205. if entry != nil {
  206. if len(entry.HardLinkId) == 0 {
  207. // only always reload hard link
  208. return entry, nil
  209. }
  210. }
  211. entry, err = file.wfs.maybeLoadEntry(file.dir.FullPath(), file.Name)
  212. if err != nil {
  213. glog.V(3).Infof("maybeLoadEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
  214. return entry, err
  215. }
  216. if entry != nil {
  217. file.setEntry(entry)
  218. } else {
  219. glog.Warningf("maybeLoadEntry not found entry %s/%s: %v", file.dir.FullPath(), file.Name, err)
  220. }
  221. return entry, nil
  222. }
  223. func lessThan(a, b *filer_pb.FileChunk) bool {
  224. if a.Mtime == b.Mtime {
  225. return a.Fid.FileKey < b.Fid.FileKey
  226. }
  227. return a.Mtime < b.Mtime
  228. }
  229. func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
  230. // find the earliest incoming chunk
  231. newChunks := chunks
  232. earliestChunk := newChunks[0]
  233. for i := 1; i < len(newChunks); i++ {
  234. if lessThan(earliestChunk, newChunks[i]) {
  235. earliestChunk = newChunks[i]
  236. }
  237. }
  238. entry := file.getEntry()
  239. if entry == nil {
  240. return
  241. }
  242. // pick out-of-order chunks from existing chunks
  243. for _, chunk := range entry.Chunks {
  244. if lessThan(earliestChunk, chunk) {
  245. chunks = append(chunks, chunk)
  246. }
  247. }
  248. // sort incoming chunks
  249. sort.Slice(chunks, func(i, j int) bool {
  250. return lessThan(chunks[i], chunks[j])
  251. })
  252. // add to entry view cache
  253. for _, chunk := range chunks {
  254. file.entryViewCache = filer.MergeIntoVisibles(file.entryViewCache, chunk)
  255. }
  256. file.reader = nil
  257. glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
  258. entry.Chunks = append(entry.Chunks, newChunks...)
  259. }
  260. func (file *File) setEntry(entry *filer_pb.Entry) {
  261. file.entryLock.Lock()
  262. defer file.entryLock.Unlock()
  263. file.entry = entry
  264. file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
  265. file.reader = nil
  266. }
  267. func (file *File) clearEntry() {
  268. file.entryLock.Lock()
  269. defer file.entryLock.Unlock()
  270. file.entry = nil
  271. file.entryViewCache = nil
  272. file.reader = nil
  273. }
  274. func (file *File) saveEntry(entry *filer_pb.Entry) error {
  275. return file.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  276. file.wfs.mapPbIdFromLocalToFiler(entry)
  277. defer file.wfs.mapPbIdFromFilerToLocal(entry)
  278. request := &filer_pb.UpdateEntryRequest{
  279. Directory: file.dir.FullPath(),
  280. Entry: entry,
  281. Signatures: []int32{file.wfs.signature},
  282. }
  283. glog.V(4).Infof("save file entry: %v", request)
  284. _, err := client.UpdateEntry(context.Background(), request)
  285. if err != nil {
  286. glog.Errorf("UpdateEntry file %s/%s: %v", file.dir.FullPath(), file.Name, err)
  287. return fuse.EIO
  288. }
  289. file.wfs.metaCache.UpdateEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
  290. return nil
  291. })
  292. }
  293. func (file *File) getEntry() *filer_pb.Entry {
  294. file.entryLock.RLock()
  295. defer file.entryLock.RUnlock()
  296. return file.entry
  297. }