weedfs_file_sync.go 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
  1. package mount
  2. import (
  3. "context"
  4. "fmt"
  5. "syscall"
  6. "time"
  7. "github.com/hanwen/go-fuse/v2/fuse"
  8. "github.com/seaweedfs/seaweedfs/weed/filer"
  9. "github.com/seaweedfs/seaweedfs/weed/glog"
  10. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  11. "github.com/seaweedfs/seaweedfs/weed/util"
  12. )
  13. /**
  14. * Flush method
  15. *
  16. * This is called on each close() of the opened file.
  17. *
  18. * Since file descriptors can be duplicated (dup, dup2, fork), for
  19. * one open call there may be many flush calls.
  20. *
  21. * Filesystems shouldn't assume that flush will always be called
  22. * after some writes, or that if will be called at all.
  23. *
  24. * fi->fh will contain the value set by the open method, or will
  25. * be undefined if the open method didn't set any value.
  26. *
  27. * NOTE: the name of the method is misleading, since (unlike
  28. * fsync) the filesystem is not forced to flush pending writes.
  29. * One reason to flush data is if the filesystem wants to return
  30. * write errors during close. However, such use is non-portable
  31. * because POSIX does not require [close] to wait for delayed I/O to
  32. * complete.
  33. *
  34. * If the filesystem supports file locking operations (setlk,
  35. * getlk) it should remove all locks belonging to 'fi->owner'.
  36. *
  37. * If this request is answered with an error code of ENOSYS,
  38. * this is treated as success and future calls to flush() will
  39. * succeed automatically without being send to the filesystem
  40. * process.
  41. *
  42. * Valid replies:
  43. * fuse_reply_err
  44. *
  45. * @param req request handle
  46. * @param ino the inode number
  47. * @param fi file information
  48. *
  49. * [close]: http://pubs.opengroup.org/onlinepubs/9699919799/functions/close.html
  50. */
  51. func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status {
  52. fh := wfs.GetHandle(FileHandleId(in.Fh))
  53. if fh == nil {
  54. return fuse.ENOENT
  55. }
  56. return wfs.doFlush(fh, in.Uid, in.Gid)
  57. }
  58. /**
  59. * Synchronize file contents
  60. *
  61. * If the datasync parameter is non-zero, then only the user data
  62. * should be flushed, not the meta data.
  63. *
  64. * If this request is answered with an error code of ENOSYS,
  65. * this is treated as success and future calls to fsync() will
  66. * succeed automatically without being send to the filesystem
  67. * process.
  68. *
  69. * Valid replies:
  70. * fuse_reply_err
  71. *
  72. * @param req request handle
  73. * @param ino the inode number
  74. * @param datasync flag indicating if only data should be flushed
  75. * @param fi file information
  76. */
  77. func (wfs *WFS) Fsync(cancel <-chan struct{}, in *fuse.FsyncIn) (code fuse.Status) {
  78. fh := wfs.GetHandle(FileHandleId(in.Fh))
  79. if fh == nil {
  80. return fuse.ENOENT
  81. }
  82. return wfs.doFlush(fh, in.Uid, in.Gid)
  83. }
  84. func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32) fuse.Status {
  85. // flush works at fh level
  86. fileFullPath := fh.FullPath()
  87. dir, name := fileFullPath.DirAndName()
  88. // send the data to the OS
  89. glog.V(4).Infof("doFlush %s fh %d", fileFullPath, fh.fh)
  90. if !wfs.IsOverQuota {
  91. if err := fh.dirtyPages.FlushData(); err != nil {
  92. glog.Errorf("%v doFlush: %v", fileFullPath, err)
  93. return fuse.EIO
  94. }
  95. }
  96. if !fh.dirtyMetadata {
  97. return fuse.OK
  98. }
  99. if wfs.IsOverQuota {
  100. return fuse.Status(syscall.ENOSPC)
  101. }
  102. fhActiveLock := fh.wfs.fhLockTable.AcquireLock("doFlush", fh.fh, util.ExclusiveLock)
  103. defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
  104. err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
  105. entry := fh.GetEntry()
  106. entry.Name = name // this flush may be just after a rename operation
  107. if entry.Attributes != nil {
  108. entry.Attributes.Mime = fh.contentType
  109. if entry.Attributes.Uid == 0 {
  110. entry.Attributes.Uid = uid
  111. }
  112. if entry.Attributes.Gid == 0 {
  113. entry.Attributes.Gid = gid
  114. }
  115. entry.Attributes.Mtime = time.Now().Unix()
  116. }
  117. request := &filer_pb.CreateEntryRequest{
  118. Directory: string(dir),
  119. Entry: entry.GetEntry(),
  120. Signatures: []int32{wfs.signature},
  121. SkipCheckParentDirectory: true,
  122. }
  123. glog.V(4).Infof("%s set chunks: %v", fileFullPath, len(entry.GetChunks()))
  124. //for i, chunk := range entry.GetChunks() {
  125. // glog.V(4).Infof("%s chunks %d: %v [%d,%d)", fileFullPath, i, chunk.GetFileIdString(), chunk.Offset, chunk.Offset+int64(chunk.Size))
  126. //}
  127. manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.GetChunks())
  128. chunks, _ := filer.CompactFileChunks(wfs.LookupFn(), nonManifestChunks)
  129. chunks, manifestErr := filer.MaybeManifestize(wfs.saveDataAsChunk(fileFullPath), chunks)
  130. if manifestErr != nil {
  131. // not good, but should be ok
  132. glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
  133. }
  134. entry.Chunks = append(chunks, manifestChunks...)
  135. wfs.mapPbIdFromLocalToFiler(request.Entry)
  136. defer wfs.mapPbIdFromFilerToLocal(request.Entry)
  137. if err := filer_pb.CreateEntry(client, request); err != nil {
  138. glog.Errorf("fh flush create %s: %v", fileFullPath, err)
  139. return fmt.Errorf("fh flush create %s: %v", fileFullPath, err)
  140. }
  141. wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, request.Entry))
  142. return nil
  143. })
  144. if err == nil {
  145. fh.dirtyMetadata = false
  146. }
  147. if err != nil {
  148. glog.Errorf("%v fh %d flush: %v", fileFullPath, fh.fh, err)
  149. return fuse.EIO
  150. }
  151. if IsDebugFileReadWrite {
  152. fh.mirrorFile.Sync()
  153. }
  154. return fuse.OK
  155. }