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.

221 lines
6.0 KiB

3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
  1. package mount
  2. import (
  3. "context"
  4. "github.com/chrislusf/seaweedfs/weed/filer"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/mount/meta_cache"
  7. "github.com/hanwen/go-fuse/v2/fuse"
  8. "math"
  9. "sync"
  10. )
  11. type DirectoryHandleId uint64
  12. type DirectoryHandle struct {
  13. isFinished bool
  14. lastEntryName string
  15. }
  16. type DirectoryHandleToInode struct {
  17. // shares the file handle id sequencer with FileHandleToInode{nextFh}
  18. sync.Mutex
  19. dir2inode map[DirectoryHandleId]*DirectoryHandle
  20. }
  21. func NewDirectoryHandleToInode() *DirectoryHandleToInode {
  22. return &DirectoryHandleToInode{
  23. dir2inode: make(map[DirectoryHandleId]*DirectoryHandle),
  24. }
  25. }
  26. func (wfs *WFS) AcquireDirectoryHandle() (DirectoryHandleId, *DirectoryHandle) {
  27. wfs.fhmap.Lock()
  28. fh := wfs.fhmap.nextFh
  29. wfs.fhmap.nextFh++
  30. wfs.fhmap.Unlock()
  31. wfs.dhmap.Lock()
  32. defer wfs.dhmap.Unlock()
  33. dh := &DirectoryHandle{
  34. isFinished: false,
  35. lastEntryName: "",
  36. }
  37. wfs.dhmap.dir2inode[DirectoryHandleId(fh)] = dh
  38. return DirectoryHandleId(fh), dh
  39. }
  40. func (wfs *WFS) GetDirectoryHandle(dhid DirectoryHandleId) *DirectoryHandle {
  41. wfs.dhmap.Lock()
  42. defer wfs.dhmap.Unlock()
  43. if dh, found := wfs.dhmap.dir2inode[dhid]; found {
  44. return dh
  45. }
  46. dh := &DirectoryHandle{
  47. isFinished: false,
  48. lastEntryName: "",
  49. }
  50. wfs.dhmap.dir2inode[dhid] = dh
  51. return dh
  52. }
  53. func (wfs *WFS) ReleaseDirectoryHandle(dhid DirectoryHandleId) {
  54. wfs.dhmap.Lock()
  55. defer wfs.dhmap.Unlock()
  56. delete(wfs.dhmap.dir2inode, dhid)
  57. }
  58. // Directory handling
  59. /** Open directory
  60. *
  61. * Unless the 'default_permissions' mount option is given,
  62. * this method should check if opendir is permitted for this
  63. * directory. Optionally opendir may also return an arbitrary
  64. * filehandle in the fuse_file_info structure, which will be
  65. * passed to readdir, releasedir and fsyncdir.
  66. */
  67. func (wfs *WFS) OpenDir(cancel <-chan struct{}, input *fuse.OpenIn, out *fuse.OpenOut) (code fuse.Status) {
  68. if !wfs.inodeToPath.HasInode(input.NodeId) {
  69. return fuse.ENOENT
  70. }
  71. dhid, _ := wfs.AcquireDirectoryHandle()
  72. out.Fh = uint64(dhid)
  73. return fuse.OK
  74. }
  75. /** Release directory
  76. *
  77. * If the directory has been removed after the call to opendir, the
  78. * path parameter will be NULL.
  79. */
  80. func (wfs *WFS) ReleaseDir(input *fuse.ReleaseIn) {
  81. wfs.ReleaseDirectoryHandle(DirectoryHandleId(input.Fh))
  82. }
  83. /** Synchronize directory contents
  84. *
  85. * If the directory has been removed after the call to opendir, the
  86. * path parameter will be NULL.
  87. *
  88. * If the datasync parameter is non-zero, then only the user data
  89. * should be flushed, not the meta data
  90. */
  91. func (wfs *WFS) FsyncDir(cancel <-chan struct{}, input *fuse.FsyncIn) (code fuse.Status) {
  92. return fuse.OK
  93. }
  94. /** Read directory
  95. *
  96. * The filesystem may choose between two modes of operation:
  97. *
  98. * 1) The readdir implementation ignores the offset parameter, and
  99. * passes zero to the filler function's offset. The filler
  100. * function will not return '1' (unless an error happens), so the
  101. * whole directory is read in a single readdir operation.
  102. *
  103. * 2) The readdir implementation keeps track of the offsets of the
  104. * directory entries. It uses the offset parameter and always
  105. * passes non-zero offset to the filler function. When the buffer
  106. * is full (or an error happens) the filler function will return
  107. * '1'.
  108. */
  109. func (wfs *WFS) ReadDir(cancel <-chan struct{}, input *fuse.ReadIn, out *fuse.DirEntryList) (code fuse.Status) {
  110. return wfs.doReadDirectory(input, out, false)
  111. }
  112. func (wfs *WFS) ReadDirPlus(cancel <-chan struct{}, input *fuse.ReadIn, out *fuse.DirEntryList) (code fuse.Status) {
  113. return wfs.doReadDirectory(input, out, true)
  114. }
  115. func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPlusMode bool) fuse.Status {
  116. dh := wfs.GetDirectoryHandle(DirectoryHandleId(input.Fh))
  117. if dh.isFinished {
  118. if input.Offset == 0 {
  119. dh.isFinished = false
  120. dh.lastEntryName = ""
  121. } else {
  122. return fuse.OK
  123. }
  124. }
  125. isEarlyTerminated := false
  126. dirPath, code := wfs.inodeToPath.GetPath(input.NodeId)
  127. if code != fuse.OK {
  128. return code
  129. }
  130. var dirEntry fuse.DirEntry
  131. if input.Offset == 0 {
  132. if !isPlusMode {
  133. out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "."})
  134. out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".."})
  135. } else {
  136. out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "."})
  137. out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".."})
  138. }
  139. }
  140. processEachEntryFn := func(entry *filer.Entry, isLast bool) bool {
  141. dirEntry.Name = entry.Name()
  142. dirEntry.Mode = toSyscallMode(entry.Mode)
  143. if !isPlusMode {
  144. inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Mode, false, entry.Inode, false)
  145. dirEntry.Ino = inode
  146. if !out.AddDirEntry(dirEntry) {
  147. isEarlyTerminated = true
  148. return false
  149. }
  150. } else {
  151. inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Mode, false, entry.Inode, true)
  152. dirEntry.Ino = inode
  153. entryOut := out.AddDirLookupEntry(dirEntry)
  154. if entryOut == nil {
  155. isEarlyTerminated = true
  156. return false
  157. }
  158. if fh, found := wfs.fhmap.FindFileHandle(inode); found {
  159. glog.V(4).Infof("readdir opened file %s", dirPath.Child(dirEntry.Name))
  160. entry = filer.FromPbEntry(string(dirPath), fh.entry)
  161. }
  162. wfs.outputFilerEntry(entryOut, inode, entry)
  163. }
  164. dh.lastEntryName = entry.Name()
  165. return true
  166. }
  167. entryChan := make(chan *filer.Entry, 128)
  168. var err error
  169. go func() {
  170. if err = meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath, entryChan); err != nil {
  171. glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
  172. }
  173. close(entryChan)
  174. }()
  175. hasData := false
  176. for entry := range entryChan {
  177. hasData = true
  178. processEachEntryFn(entry, false)
  179. }
  180. if err != nil {
  181. return fuse.EIO
  182. }
  183. if !hasData {
  184. listErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, dh.lastEntryName, false, int64(math.MaxInt32), func(entry *filer.Entry) bool {
  185. return processEachEntryFn(entry, false)
  186. })
  187. if listErr != nil {
  188. glog.Errorf("list meta cache: %v", listErr)
  189. return fuse.EIO
  190. }
  191. }
  192. if !isEarlyTerminated {
  193. dh.isFinished = true
  194. }
  195. return fuse.OK
  196. }