aboutsummaryrefslogtreecommitdiff
path: root/weed/filesys/wfs.go
diff options
context:
space:
mode:
Diffstat (limited to 'weed/filesys/wfs.go')
-rw-r--r--weed/filesys/wfs.go133
1 files changed, 67 insertions, 66 deletions
diff --git a/weed/filesys/wfs.go b/weed/filesys/wfs.go
index 4807e367b..67dd2a62c 100644
--- a/weed/filesys/wfs.go
+++ b/weed/filesys/wfs.go
@@ -5,17 +5,21 @@ import (
"fmt"
"math"
"os"
+ "path"
"strings"
"sync"
"time"
+ "github.com/chrislusf/seaweedfs/weed/util/grace"
"github.com/karlseguin/ccache"
"google.golang.org/grpc"
- "github.com/chrislusf/seaweedfs/weed/filer2"
+ "github.com/chrislusf/seaweedfs/weed/filesys/meta_cache"
"github.com/chrislusf/seaweedfs/weed/glog"
+ "github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
+ "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
"github.com/seaweedfs/fuse"
"github.com/seaweedfs/fuse/fs"
)
@@ -28,6 +32,8 @@ type Option struct {
Replication string
TtlSec int32
ChunkSizeLimit int64
+ CacheDir string
+ CacheSizeMB int64
DataCenter string
DirListCacheLimit int64
EntryCacheTtl time.Duration
@@ -38,6 +44,11 @@ type Option struct {
MountMode os.FileMode
MountCtime time.Time
MountMtime time.Time
+
+ OutsideContainerClusterMode bool // whether the mount runs outside SeaweedFS containers
+ Cipher bool // whether encrypt data on volume server
+ AsyncMetaDataCaching bool // whether asynchronously cache meta data
+
}
var _ = fs.FS(&WFS{})
@@ -48,18 +59,18 @@ type WFS struct {
listDirectoryEntriesCache *ccache.Cache
// contains all open handles, protected by handlesLock
- handlesLock sync.Mutex
- handles []*FileHandle
- pathToHandleIndex map[filer2.FullPath]int
+ handlesLock sync.Mutex
+ handles map[uint64]*FileHandle
bufPool sync.Pool
stats statsCache
- // nodes, protected by nodesLock
- nodesLock sync.Mutex
- nodes map[uint64]fs.Node
- root fs.Node
+ root fs.Node
+ fsNodeCache *FsCache
+
+ chunkCache *chunk_cache.ChunkCache
+ metaCache *meta_cache.MetaCache
}
type statsCache struct {
filer_pb.StatisticsResponse
@@ -70,16 +81,34 @@ func NewSeaweedFileSystem(option *Option) *WFS {
wfs := &WFS{
option: option,
listDirectoryEntriesCache: ccache.New(ccache.Configure().MaxSize(option.DirListCacheLimit * 3).ItemsToPrune(100)),
- pathToHandleIndex: make(map[filer2.FullPath]int),
+ handles: make(map[uint64]*FileHandle),
bufPool: sync.Pool{
New: func() interface{} {
return make([]byte, option.ChunkSizeLimit)
},
},
- nodes: make(map[uint64]fs.Node),
+ }
+ if option.CacheSizeMB > 0 {
+ wfs.chunkCache = chunk_cache.NewChunkCache(256, option.CacheDir, option.CacheSizeMB)
+ grace.OnInterrupt(func() {
+ wfs.chunkCache.Shutdown()
+ })
+ }
+ if wfs.option.AsyncMetaDataCaching {
+ wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.CacheDir, "meta"))
+ startTime := time.Now()
+ if err := meta_cache.InitMetaCache(wfs.metaCache, wfs, wfs.option.FilerMountRootPath); err != nil {
+ glog.V(0).Infof("failed to init meta cache: %v", err)
+ } else {
+ go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
+ grace.OnInterrupt(func() {
+ wfs.metaCache.Shutdown()
+ })
+ }
}
- wfs.root = &Dir{Path: wfs.option.FilerMountRootPath, wfs: wfs}
+ wfs.root = &Dir{name: wfs.option.FilerMountRootPath, wfs: wfs}
+ wfs.fsNodeCache = newFsCache(wfs.root)
return wfs
}
@@ -88,23 +117,18 @@ func (wfs *WFS) Root() (fs.Node, error) {
return wfs.root, nil
}
-func (wfs *WFS) WithFilerClient(ctx context.Context, fn func(context.Context, filer_pb.SeaweedFilerClient) error) error {
+var _ = filer_pb.FilerClient(&WFS{})
- err := util.WithCachedGrpcClient(ctx, func(ctx2 context.Context, grpcConnection *grpc.ClientConn) error {
+func (wfs *WFS) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) error {
+
+ err := pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
client := filer_pb.NewSeaweedFilerClient(grpcConnection)
- return fn(ctx2, client)
+ return fn(client)
}, wfs.option.FilerGrpcAddress, wfs.option.GrpcDialOption)
if err == nil {
return nil
}
- if strings.Contains(err.Error(), "context canceled") {
- glog.V(2).Infoln("retry context canceled request...")
- return util.WithCachedGrpcClient(context.Background(), func(ctx2 context.Context, grpcConnection *grpc.ClientConn) error {
- client := filer_pb.NewSeaweedFilerClient(grpcConnection)
- return fn(ctx2, client)
- }, wfs.option.FilerGrpcAddress, wfs.option.GrpcDialOption)
- }
return err
}
@@ -117,40 +141,27 @@ func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHand
wfs.handlesLock.Lock()
defer wfs.handlesLock.Unlock()
- index, found := wfs.pathToHandleIndex[fullpath]
- if found && wfs.handles[index] != nil {
- glog.V(2).Infoln(fullpath, "found fileHandle id", index)
- return wfs.handles[index]
+ inodeId := file.fullpath().AsInode()
+ existingHandle, found := wfs.handles[inodeId]
+ if found && existingHandle != nil {
+ return existingHandle
}
fileHandle = newFileHandle(file, uid, gid)
- for i, h := range wfs.handles {
- if h == nil {
- wfs.handles[i] = fileHandle
- fileHandle.handle = uint64(i)
- wfs.pathToHandleIndex[fullpath] = i
- glog.V(4).Infof("%s reuse fh %d", fullpath, fileHandle.handle)
- return
- }
- }
-
- wfs.handles = append(wfs.handles, fileHandle)
- fileHandle.handle = uint64(len(wfs.handles) - 1)
- wfs.pathToHandleIndex[fullpath] = int(fileHandle.handle)
+ wfs.handles[inodeId] = fileHandle
+ fileHandle.handle = inodeId
glog.V(4).Infof("%s new fh %d", fullpath, fileHandle.handle)
return
}
-func (wfs *WFS) ReleaseHandle(fullpath filer2.FullPath, handleId fuse.HandleID) {
+func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
wfs.handlesLock.Lock()
defer wfs.handlesLock.Unlock()
glog.V(4).Infof("%s ReleaseHandle id %d current handles length %d", fullpath, handleId, len(wfs.handles))
- delete(wfs.pathToHandleIndex, fullpath)
- if int(handleId) < len(wfs.handles) {
- wfs.handles[int(handleId)] = nil
- }
+
+ delete(wfs.handles, fullpath.AsInode())
return
}
@@ -162,7 +173,7 @@ func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.
if wfs.stats.lastChecked < time.Now().Unix()-20 {
- err := wfs.WithFilerClient(ctx, func(ctx context.Context, client filer_pb.SeaweedFilerClient) error {
+ err := wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.StatisticsRequest{
Collection: wfs.option.Collection,
@@ -171,7 +182,7 @@ func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.
}
glog.V(4).Infof("reading filer stats: %+v", request)
- resp, err := client.Statistics(ctx, request)
+ resp, err := client.Statistics(context.Background(), request)
if err != nil {
glog.V(0).Infof("reading filer stats %v: %v", request, err)
return err
@@ -217,43 +228,33 @@ func (wfs *WFS) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse.
return nil
}
-func (wfs *WFS) cacheGet(path filer2.FullPath) *filer_pb.Entry {
+func (wfs *WFS) cacheGet(path util.FullPath) *filer_pb.Entry {
item := wfs.listDirectoryEntriesCache.Get(string(path))
if item != nil && !item.Expired() {
return item.Value().(*filer_pb.Entry)
}
return nil
}
-func (wfs *WFS) cacheSet(path filer2.FullPath, entry *filer_pb.Entry, ttl time.Duration) {
+func (wfs *WFS) cacheSet(path util.FullPath, entry *filer_pb.Entry, ttl time.Duration) {
if entry == nil {
wfs.listDirectoryEntriesCache.Delete(string(path))
} else {
wfs.listDirectoryEntriesCache.Set(string(path), entry, ttl)
}
}
-func (wfs *WFS) cacheDelete(path filer2.FullPath) {
+func (wfs *WFS) cacheDelete(path util.FullPath) {
wfs.listDirectoryEntriesCache.Delete(string(path))
}
-func (wfs *WFS) getNode(fullpath filer2.FullPath, fn func() fs.Node) fs.Node {
- wfs.nodesLock.Lock()
- defer wfs.nodesLock.Unlock()
-
- node, found := wfs.nodes[fullpath.AsInode()]
- if found {
- return node
+func (wfs *WFS) AdjustedUrl(hostAndPort string) string {
+ if !wfs.option.OutsideContainerClusterMode {
+ return hostAndPort
}
- node = fn()
- if node != nil {
- wfs.nodes[fullpath.AsInode()] = node
+ commaIndex := strings.Index(hostAndPort, ":")
+ if commaIndex < 0 {
+ return hostAndPort
}
- return node
-}
-
-func (wfs *WFS) forgetNode(fullpath filer2.FullPath) {
- wfs.nodesLock.Lock()
- defer wfs.nodesLock.Unlock()
-
- delete(wfs.nodes, fullpath.AsInode())
+ filerCommaIndex := strings.Index(wfs.option.FilerGrpcAddress, ":")
+ return fmt.Sprintf("%s:%s", wfs.option.FilerGrpcAddress[:filerCommaIndex], hostAndPort[commaIndex+1:])
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-23 23:14:23 +0000