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.

224 lines
6.0 KiB

7 years ago
5 years ago
5 years ago
6 years ago
4 years ago
5 years ago
4 years ago
5 years ago
4 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "fmt"
  5. "math"
  6. "os"
  7. "path"
  8. "sync"
  9. "time"
  10. "google.golang.org/grpc"
  11. "github.com/chrislusf/seaweedfs/weed/util/grace"
  12. "github.com/seaweedfs/fuse"
  13. "github.com/seaweedfs/fuse/fs"
  14. "github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
  15. "github.com/chrislusf/seaweedfs/weed/glog"
  16. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  17. "github.com/chrislusf/seaweedfs/weed/util"
  18. "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
  19. )
  20. type Option struct {
  21. FilerGrpcAddress string
  22. GrpcDialOption grpc.DialOption
  23. FilerMountRootPath string
  24. Collection string
  25. Replication string
  26. TtlSec int32
  27. ChunkSizeLimit int64
  28. CacheDir string
  29. CacheSizeMB int64
  30. DataCenter string
  31. EntryCacheTtl time.Duration
  32. Umask os.FileMode
  33. MountUid uint32
  34. MountGid uint32
  35. MountMode os.FileMode
  36. MountCtime time.Time
  37. MountMtime time.Time
  38. OutsideContainerClusterMode bool // whether the mount runs outside SeaweedFS containers
  39. Cipher bool // whether encrypt data on volume server
  40. UidGidMapper *meta_cache.UidGidMapper
  41. }
  42. var _ = fs.FS(&WFS{})
  43. var _ = fs.FSStatfser(&WFS{})
  44. type WFS struct {
  45. option *Option
  46. // contains all open handles, protected by handlesLock
  47. handlesLock sync.Mutex
  48. handles map[uint64]*FileHandle
  49. bufPool sync.Pool
  50. stats statsCache
  51. root fs.Node
  52. fsNodeCache *FsCache
  53. chunkCache *chunk_cache.TieredChunkCache
  54. metaCache *meta_cache.MetaCache
  55. signature int32
  56. }
  57. type statsCache struct {
  58. filer_pb.StatisticsResponse
  59. lastChecked int64 // unix time in seconds
  60. }
  61. func NewSeaweedFileSystem(option *Option) *WFS {
  62. wfs := &WFS{
  63. option: option,
  64. handles: make(map[uint64]*FileHandle),
  65. bufPool: sync.Pool{
  66. New: func() interface{} {
  67. return make([]byte, option.ChunkSizeLimit)
  68. },
  69. },
  70. signature: util.RandomInt32(),
  71. }
  72. cacheUniqueId := util.Md5String([]byte(option.FilerGrpcAddress + option.FilerMountRootPath + util.Version()))[0:4]
  73. cacheDir := path.Join(option.CacheDir, cacheUniqueId)
  74. if option.CacheSizeMB > 0 {
  75. os.MkdirAll(cacheDir, os.FileMode(0777)&^option.Umask)
  76. wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, cacheDir, option.CacheSizeMB, 1024*1024)
  77. }
  78. wfs.metaCache = meta_cache.NewMetaCache(path.Join(cacheDir, "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper)
  79. startTime := time.Now()
  80. go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
  81. grace.OnInterrupt(func() {
  82. wfs.metaCache.Shutdown()
  83. })
  84. entry, _ := filer_pb.GetEntry(wfs, util.FullPath(wfs.option.FilerMountRootPath))
  85. wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs, entry: entry}
  86. wfs.fsNodeCache = newFsCache(wfs.root)
  87. return wfs
  88. }
  89. func (wfs *WFS) Root() (fs.Node, error) {
  90. return wfs.root, nil
  91. }
  92. func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHandle) {
  93. fullpath := file.fullpath()
  94. glog.V(4).Infof("AcquireHandle %s uid=%d gid=%d", fullpath, uid, gid)
  95. wfs.handlesLock.Lock()
  96. defer wfs.handlesLock.Unlock()
  97. inodeId := file.fullpath().AsInode()
  98. if file.isOpen > 0 {
  99. existingHandle, found := wfs.handles[inodeId]
  100. if found && existingHandle != nil {
  101. file.isOpen++
  102. return existingHandle
  103. }
  104. }
  105. fileHandle = newFileHandle(file, uid, gid)
  106. file.maybeLoadEntry(context.Background())
  107. file.isOpen++
  108. wfs.handles[inodeId] = fileHandle
  109. fileHandle.handle = inodeId
  110. return
  111. }
  112. func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
  113. wfs.handlesLock.Lock()
  114. defer wfs.handlesLock.Unlock()
  115. glog.V(4).Infof("%s ReleaseHandle id %d current handles length %d", fullpath, handleId, len(wfs.handles))
  116. delete(wfs.handles, fullpath.AsInode())
  117. return
  118. }
  119. // Statfs is called to obtain file system metadata. Implements fuse.FSStatfser
  120. func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.StatfsResponse) error {
  121. glog.V(4).Infof("reading fs stats: %+v", req)
  122. if wfs.stats.lastChecked < time.Now().Unix()-20 {
  123. err := wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  124. request := &filer_pb.StatisticsRequest{
  125. Collection: wfs.option.Collection,
  126. Replication: wfs.option.Replication,
  127. Ttl: fmt.Sprintf("%ds", wfs.option.TtlSec),
  128. }
  129. glog.V(4).Infof("reading filer stats: %+v", request)
  130. resp, err := client.Statistics(context.Background(), request)
  131. if err != nil {
  132. glog.V(0).Infof("reading filer stats %v: %v", request, err)
  133. return err
  134. }
  135. glog.V(4).Infof("read filer stats: %+v", resp)
  136. wfs.stats.TotalSize = resp.TotalSize
  137. wfs.stats.UsedSize = resp.UsedSize
  138. wfs.stats.FileCount = resp.FileCount
  139. wfs.stats.lastChecked = time.Now().Unix()
  140. return nil
  141. })
  142. if err != nil {
  143. glog.V(0).Infof("filer Statistics: %v", err)
  144. return err
  145. }
  146. }
  147. totalDiskSize := wfs.stats.TotalSize
  148. usedDiskSize := wfs.stats.UsedSize
  149. actualFileCount := wfs.stats.FileCount
  150. // Compute the total number of available blocks
  151. resp.Blocks = totalDiskSize / blockSize
  152. // Compute the number of used blocks
  153. numBlocks := uint64(usedDiskSize / blockSize)
  154. // Report the number of free and available blocks for the block size
  155. resp.Bfree = resp.Blocks - numBlocks
  156. resp.Bavail = resp.Blocks - numBlocks
  157. resp.Bsize = uint32(blockSize)
  158. // Report the total number of possible files in the file system (and those free)
  159. resp.Files = math.MaxInt64
  160. resp.Ffree = math.MaxInt64 - actualFileCount
  161. // Report the maximum length of a name and the minimum fragment size
  162. resp.Namelen = 1024
  163. resp.Frsize = uint32(blockSize)
  164. return nil
  165. }
  166. func (wfs *WFS) mapPbIdFromFilerToLocal(entry *filer_pb.Entry) {
  167. if entry.Attributes == nil {
  168. return
  169. }
  170. entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.FilerToLocal(entry.Attributes.Uid, entry.Attributes.Gid)
  171. }
  172. func (wfs *WFS) mapPbIdFromLocalToFiler(entry *filer_pb.Entry) {
  173. if entry.Attributes == nil {
  174. return
  175. }
  176. entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.LocalToFiler(entry.Attributes.Uid, entry.Attributes.Gid)
  177. }