|
@ -9,8 +9,6 @@ import ( |
|
|
|
|
|
|
|
|
"google.golang.org/grpc" |
|
|
"google.golang.org/grpc" |
|
|
|
|
|
|
|
|
"github.com/karlseguin/ccache" |
|
|
|
|
|
|
|
|
|
|
|
"github.com/chrislusf/seaweedfs/weed/glog" |
|
|
"github.com/chrislusf/seaweedfs/weed/glog" |
|
|
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" |
|
|
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" |
|
|
"github.com/chrislusf/seaweedfs/weed/util" |
|
|
"github.com/chrislusf/seaweedfs/weed/util" |
|
@ -27,7 +25,6 @@ var ( |
|
|
|
|
|
|
|
|
type Filer struct { |
|
|
type Filer struct { |
|
|
Store *FilerStoreWrapper |
|
|
Store *FilerStoreWrapper |
|
|
directoryCache *ccache.Cache |
|
|
|
|
|
MasterClient *wdclient.MasterClient |
|
|
MasterClient *wdclient.MasterClient |
|
|
fileIdDeletionQueue *util.UnboundedQueue |
|
|
fileIdDeletionQueue *util.UnboundedQueue |
|
|
GrpcDialOption grpc.DialOption |
|
|
GrpcDialOption grpc.DialOption |
|
@ -44,7 +41,6 @@ type Filer struct { |
|
|
func NewFiler(masters []string, grpcDialOption grpc.DialOption, |
|
|
func NewFiler(masters []string, grpcDialOption grpc.DialOption, |
|
|
filerHost string, filerGrpcPort uint32, collection string, replication string, notifyFn func()) *Filer { |
|
|
filerHost string, filerGrpcPort uint32, collection string, replication string, notifyFn func()) *Filer { |
|
|
f := &Filer{ |
|
|
f := &Filer{ |
|
|
directoryCache: ccache.New(ccache.Configure().MaxSize(1000).ItemsToPrune(100)), |
|
|
|
|
|
MasterClient: wdclient.NewMasterClient(grpcDialOption, "filer", filerHost, filerGrpcPort, masters), |
|
|
MasterClient: wdclient.NewMasterClient(grpcDialOption, "filer", filerHost, filerGrpcPort, masters), |
|
|
fileIdDeletionQueue: util.NewUnboundedQueue(), |
|
|
fileIdDeletionQueue: util.NewUnboundedQueue(), |
|
|
GrpcDialOption: grpcDialOption, |
|
|
GrpcDialOption: grpcDialOption, |
|
@ -77,10 +73,6 @@ func (f *Filer) GetStore() (store FilerStore) { |
|
|
return f.Store |
|
|
return f.Store |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
func (f *Filer) DisableDirectoryCache() { |
|
|
|
|
|
f.directoryCache = nil |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (fs *Filer) GetMaster() string { |
|
|
func (fs *Filer) GetMaster() string { |
|
|
return fs.MasterClient.GetMaster() |
|
|
return fs.MasterClient.GetMaster() |
|
|
} |
|
|
} |
|
@ -117,16 +109,9 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, o_excl bool, isFr |
|
|
dirPath := "/" + util.Join(dirParts[:i]...) |
|
|
dirPath := "/" + util.Join(dirParts[:i]...) |
|
|
// fmt.Printf("%d directory: %+v\n", i, dirPath)
|
|
|
// fmt.Printf("%d directory: %+v\n", i, dirPath)
|
|
|
|
|
|
|
|
|
// first check local cache
|
|
|
|
|
|
dirEntry := f.cacheGetDirectory(dirPath) |
|
|
|
|
|
|
|
|
|
|
|
// not found, check the store directly
|
|
|
|
|
|
if dirEntry == nil { |
|
|
|
|
|
|
|
|
// check the store directly, skipping cached directories
|
|
|
glog.V(4).Infof("find uncached directory: %s", dirPath) |
|
|
glog.V(4).Infof("find uncached directory: %s", dirPath) |
|
|
dirEntry, _ = f.FindEntry(ctx, util.FullPath(dirPath)) |
|
|
|
|
|
} else { |
|
|
|
|
|
// glog.V(4).Infof("found cached directory: %s", dirPath)
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
dirEntry, _ := f.FindEntry(ctx, util.FullPath(dirPath)) |
|
|
|
|
|
|
|
|
// no such existing directory
|
|
|
// no such existing directory
|
|
|
if dirEntry == nil { |
|
|
if dirEntry == nil { |
|
@ -166,9 +151,6 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, o_excl bool, isFr |
|
|
return fmt.Errorf("%s is a file", dirPath) |
|
|
return fmt.Errorf("%s is a file", dirPath) |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
// cache the directory entry
|
|
|
|
|
|
f.cacheSetDirectory(dirPath, dirEntry, i) |
|
|
|
|
|
|
|
|
|
|
|
// remember the direct parent directory entry
|
|
|
// remember the direct parent directory entry
|
|
|
if i == len(dirParts)-1 { |
|
|
if i == len(dirParts)-1 { |
|
|
lastDirectoryEntry = dirEntry |
|
|
lastDirectoryEntry = dirEntry |
|
@ -295,45 +277,6 @@ func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, sta |
|
|
return |
|
|
return |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
func (f *Filer) cacheDelDirectory(dirpath string) { |
|
|
|
|
|
|
|
|
|
|
|
if dirpath == "/" { |
|
|
|
|
|
return |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
if f.directoryCache == nil { |
|
|
|
|
|
return |
|
|
|
|
|
} |
|
|
|
|
|
f.directoryCache.Delete(dirpath) |
|
|
|
|
|
return |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (f *Filer) cacheGetDirectory(dirpath string) *Entry { |
|
|
|
|
|
|
|
|
|
|
|
if f.directoryCache == nil { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
item := f.directoryCache.Get(dirpath) |
|
|
|
|
|
if item == nil { |
|
|
|
|
|
return nil |
|
|
|
|
|
} |
|
|
|
|
|
return item.Value().(*Entry) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (f *Filer) cacheSetDirectory(dirpath string, dirEntry *Entry, level int) { |
|
|
|
|
|
|
|
|
|
|
|
if f.directoryCache == nil { |
|
|
|
|
|
return |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
minutes := 60 |
|
|
|
|
|
if level < 10 { |
|
|
|
|
|
minutes -= level * 6 |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
f.directoryCache.Set(dirpath, dirEntry, time.Duration(minutes)*time.Minute) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (f *Filer) Shutdown() { |
|
|
func (f *Filer) Shutdown() { |
|
|
f.LocalMetaLogBuffer.Shutdown() |
|
|
f.LocalMetaLogBuffer.Shutdown() |
|
|
f.Store.Shutdown() |
|
|
f.Store.Shutdown() |
|
|