weedfs_dir_read.go 6.8 KB

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