weedfs_dir_read.go 6.8 KB

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