Merge remote-tracking branch 'upstream/master'

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:])
 
 }