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.

303 lines
8.5 KiB

7 years ago
5 years ago
5 years ago
4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
5 years ago
4 years ago
4 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/pb"
  6. "math"
  7. "math/rand"
  8. "os"
  9. "path"
  10. "path/filepath"
  11. "sync"
  12. "time"
  13. "github.com/chrislusf/seaweedfs/weed/filer"
  14. "github.com/chrislusf/seaweedfs/weed/storage/types"
  15. "github.com/chrislusf/seaweedfs/weed/wdclient"
  16. "google.golang.org/grpc"
  17. "github.com/chrislusf/seaweedfs/weed/util/grace"
  18. "github.com/seaweedfs/fuse"
  19. "github.com/seaweedfs/fuse/fs"
  20. "github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
  21. "github.com/chrislusf/seaweedfs/weed/glog"
  22. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  23. "github.com/chrislusf/seaweedfs/weed/util"
  24. "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
  25. )
  26. type Option struct {
  27. MountDirectory string
  28. FilerAddresses []pb.ServerAddress
  29. filerIndex int
  30. GrpcDialOption grpc.DialOption
  31. FilerMountRootPath string
  32. Collection string
  33. Replication string
  34. TtlSec int32
  35. DiskType types.DiskType
  36. ChunkSizeLimit int64
  37. ConcurrentWriters int
  38. CacheDir string
  39. CacheSizeMB int64
  40. DataCenter string
  41. Umask os.FileMode
  42. MountUid uint32
  43. MountGid uint32
  44. MountMode os.FileMode
  45. MountCtime time.Time
  46. MountMtime time.Time
  47. MountParentInode uint64
  48. VolumeServerAccess string // how to access volume servers
  49. Cipher bool // whether encrypt data on volume server
  50. UidGidMapper *meta_cache.UidGidMapper
  51. uniqueCacheDir string
  52. uniqueCacheTempPageDir string
  53. }
  54. var _ = fs.FS(&WFS{})
  55. var _ = fs.FSStatfser(&WFS{})
  56. type WFS struct {
  57. option *Option
  58. // contains all open handles, protected by handlesLock
  59. handlesLock sync.Mutex
  60. handles map[uint64]*FileHandle
  61. bufPool sync.Pool
  62. stats statsCache
  63. root fs.Node
  64. fsNodeCache *FsCache
  65. chunkCache *chunk_cache.TieredChunkCache
  66. metaCache *meta_cache.MetaCache
  67. signature int32
  68. // throttle writers
  69. concurrentWriters *util.LimitedConcurrentExecutor
  70. Server *fs.Server
  71. }
  72. type statsCache struct {
  73. filer_pb.StatisticsResponse
  74. lastChecked int64 // unix time in seconds
  75. }
  76. func NewSeaweedFileSystem(option *Option) *WFS {
  77. wfs := &WFS{
  78. option: option,
  79. handles: make(map[uint64]*FileHandle),
  80. bufPool: sync.Pool{
  81. New: func() interface{} {
  82. return make([]byte, option.ChunkSizeLimit)
  83. },
  84. },
  85. signature: util.RandomInt32(),
  86. }
  87. wfs.option.filerIndex = rand.Intn(len(option.FilerAddresses))
  88. wfs.option.setupUniqueCacheDirectory()
  89. if option.CacheSizeMB > 0 {
  90. wfs.chunkCache = chunk_cache.NewTieredChunkCache(256, option.getUniqueCacheDir(), option.CacheSizeMB, 1024*1024)
  91. }
  92. wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.getUniqueCacheDir(), "meta"), util.FullPath(option.FilerMountRootPath), option.UidGidMapper, func(filePath util.FullPath) {
  93. fsNode := NodeWithId(filePath.AsInode())
  94. if err := wfs.Server.InvalidateNodeData(fsNode); err != nil {
  95. glog.V(4).Infof("InvalidateNodeData %s : %v", filePath, err)
  96. }
  97. dir, name := filePath.DirAndName()
  98. parent := NodeWithId(util.FullPath(dir).AsInode())
  99. if dir == option.FilerMountRootPath {
  100. parent = NodeWithId(1)
  101. }
  102. if err := wfs.Server.InvalidateEntry(parent, name); err != nil {
  103. glog.V(4).Infof("InvalidateEntry %s : %v", filePath, err)
  104. }
  105. })
  106. grace.OnInterrupt(func() {
  107. wfs.metaCache.Shutdown()
  108. })
  109. wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs, id: 1}
  110. wfs.fsNodeCache = newFsCache(wfs.root)
  111. if wfs.option.ConcurrentWriters > 0 {
  112. wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(wfs.option.ConcurrentWriters)
  113. }
  114. return wfs
  115. }
  116. func (wfs *WFS) StartBackgroundTasks() {
  117. startTime := time.Now()
  118. go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
  119. }
  120. func (wfs *WFS) Root() (fs.Node, error) {
  121. return wfs.root, nil
  122. }
  123. func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32, writeOnly bool) (fileHandle *FileHandle) {
  124. fullpath := file.fullpath()
  125. glog.V(4).Infof("AcquireHandle %s uid=%d gid=%d", fullpath, uid, gid)
  126. inodeId := file.Id()
  127. wfs.handlesLock.Lock()
  128. existingHandle, found := wfs.handles[inodeId]
  129. if found && existingHandle != nil && existingHandle.f.isOpen > 0 {
  130. existingHandle.f.isOpen++
  131. wfs.handlesLock.Unlock()
  132. existingHandle.dirtyPages.SetWriteOnly(writeOnly)
  133. glog.V(4).Infof("Reuse AcquiredHandle %s open %d", fullpath, existingHandle.f.isOpen)
  134. return existingHandle
  135. }
  136. wfs.handlesLock.Unlock()
  137. entry, _ := file.maybeLoadEntry(context.Background())
  138. file.entry = entry
  139. fileHandle = newFileHandle(file, uid, gid, writeOnly)
  140. wfs.handlesLock.Lock()
  141. file.isOpen++
  142. wfs.handles[inodeId] = fileHandle
  143. wfs.handlesLock.Unlock()
  144. fileHandle.handle = inodeId
  145. glog.V(4).Infof("Acquired new Handle %s open %d", fullpath, file.isOpen)
  146. return
  147. }
  148. func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
  149. wfs.handlesLock.Lock()
  150. defer wfs.handlesLock.Unlock()
  151. glog.V(4).Infof("ReleaseHandle %s id %d current handles length %d", fullpath, handleId, len(wfs.handles))
  152. delete(wfs.handles, uint64(handleId))
  153. return
  154. }
  155. // Statfs is called to obtain file system metadata. Implements fuse.FSStatfser
  156. func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.StatfsResponse) error {
  157. glog.V(4).Infof("reading fs stats: %+v", req)
  158. if wfs.stats.lastChecked < time.Now().Unix()-20 {
  159. err := wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  160. request := &filer_pb.StatisticsRequest{
  161. Collection: wfs.option.Collection,
  162. Replication: wfs.option.Replication,
  163. Ttl: fmt.Sprintf("%ds", wfs.option.TtlSec),
  164. DiskType: string(wfs.option.DiskType),
  165. }
  166. glog.V(4).Infof("reading filer stats: %+v", request)
  167. resp, err := client.Statistics(context.Background(), request)
  168. if err != nil {
  169. glog.V(0).Infof("reading filer stats %v: %v", request, err)
  170. return err
  171. }
  172. glog.V(4).Infof("read filer stats: %+v", resp)
  173. wfs.stats.TotalSize = resp.TotalSize
  174. wfs.stats.UsedSize = resp.UsedSize
  175. wfs.stats.FileCount = resp.FileCount
  176. wfs.stats.lastChecked = time.Now().Unix()
  177. return nil
  178. })
  179. if err != nil {
  180. glog.V(0).Infof("filer Statistics: %v", err)
  181. return err
  182. }
  183. }
  184. totalDiskSize := wfs.stats.TotalSize
  185. usedDiskSize := wfs.stats.UsedSize
  186. actualFileCount := wfs.stats.FileCount
  187. // Compute the total number of available blocks
  188. resp.Blocks = totalDiskSize / blockSize
  189. // Compute the number of used blocks
  190. numBlocks := uint64(usedDiskSize / blockSize)
  191. // Report the number of free and available blocks for the block size
  192. resp.Bfree = resp.Blocks - numBlocks
  193. resp.Bavail = resp.Blocks - numBlocks
  194. resp.Bsize = uint32(blockSize)
  195. // Report the total number of possible files in the file system (and those free)
  196. resp.Files = math.MaxInt64
  197. resp.Ffree = math.MaxInt64 - actualFileCount
  198. // Report the maximum length of a name and the minimum fragment size
  199. resp.Namelen = 1024
  200. resp.Frsize = uint32(blockSize)
  201. return nil
  202. }
  203. func (wfs *WFS) mapPbIdFromFilerToLocal(entry *filer_pb.Entry) {
  204. if entry.Attributes == nil {
  205. return
  206. }
  207. entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.FilerToLocal(entry.Attributes.Uid, entry.Attributes.Gid)
  208. }
  209. func (wfs *WFS) mapPbIdFromLocalToFiler(entry *filer_pb.Entry) {
  210. if entry.Attributes == nil {
  211. return
  212. }
  213. entry.Attributes.Uid, entry.Attributes.Gid = wfs.option.UidGidMapper.LocalToFiler(entry.Attributes.Uid, entry.Attributes.Gid)
  214. }
  215. func (wfs *WFS) LookupFn() wdclient.LookupFileIdFunctionType {
  216. if wfs.option.VolumeServerAccess == "filerProxy" {
  217. return func(fileId string) (targetUrls []string, err error) {
  218. return []string{"http://" + wfs.getCurrentFiler().ToHttpAddress() + "/?proxyChunkId=" + fileId}, nil
  219. }
  220. }
  221. return filer.LookupFn(wfs)
  222. }
  223. func (wfs *WFS) getCurrentFiler() pb.ServerAddress {
  224. return wfs.option.FilerAddresses[wfs.option.filerIndex]
  225. }
  226. func (option *Option) setupUniqueCacheDirectory() {
  227. cacheUniqueId := util.Md5String([]byte(option.MountDirectory + string(option.FilerAddresses[0]) + option.FilerMountRootPath + util.Version()))[0:8]
  228. option.uniqueCacheDir = path.Join(option.CacheDir, cacheUniqueId)
  229. option.uniqueCacheTempPageDir = filepath.Join(option.uniqueCacheDir, "sw")
  230. os.MkdirAll(option.uniqueCacheTempPageDir, os.FileMode(0777)&^option.Umask)
  231. }
  232. func (option *Option) getTempFilePageDir() string {
  233. return option.uniqueCacheTempPageDir
  234. }
  235. func (option *Option) getUniqueCacheDir() string {
  236. return option.uniqueCacheDir
  237. }
  238. type NodeWithId uint64
  239. func (n NodeWithId) Id() uint64 {
  240. return uint64(n)
  241. }
  242. func (n NodeWithId) Attr(ctx context.Context, attr *fuse.Attr) error {
  243. return nil
  244. }