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.

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