seaweedfs/weed/filesys/meta_cache/meta_cache.go

147 lines
3.9 KiB
Go
Raw Normal View History

package meta_cache
import (
2020-04-30 01:20:54 +00:00
"context"
"fmt"
"os"
2020-04-30 01:20:54 +00:00
"sync"
2020-09-01 07:21:19 +00:00
"github.com/chrislusf/seaweedfs/weed/filer"
"github.com/chrislusf/seaweedfs/weed/filer/leveldb"
"github.com/chrislusf/seaweedfs/weed/glog"
2020-04-30 01:20:54 +00:00
"github.com/chrislusf/seaweedfs/weed/util"
2020-06-19 16:45:27 +00:00
"github.com/chrislusf/seaweedfs/weed/util/bounded_tree"
)
// need to have logic similar to FilerStoreWrapper
// e.g. fill fileId field for chunks
type MetaCache struct {
2020-09-24 10:06:44 +00:00
localStore filer.VirtualFilerStore
2020-04-30 01:20:54 +00:00
sync.RWMutex
2020-06-19 16:45:27 +00:00
visitedBoundary *bounded_tree.BoundedTree
uidGidMapper *UidGidMapper
invalidateFunc func(util.FullPath)
}
func NewMetaCache(dbFolder string, baseDir util.FullPath, uidGidMapper *UidGidMapper, invalidateFunc func(util.FullPath)) *MetaCache {
return &MetaCache{
2020-09-22 23:24:13 +00:00
localStore: openMetaStore(dbFolder),
visitedBoundary: bounded_tree.NewBoundedTree(baseDir),
uidGidMapper: uidGidMapper,
invalidateFunc: invalidateFunc,
}
}
2020-09-24 10:06:44 +00:00
func openMetaStore(dbFolder string) filer.VirtualFilerStore {
2020-04-22 05:00:34 +00:00
os.RemoveAll(dbFolder)
os.MkdirAll(dbFolder, 0755)
store := &leveldb.LevelDBStore{}
2020-04-22 05:00:34 +00:00
config := &cacheConfig{
dir: dbFolder,
}
if err := store.Initialize(config, ""); err != nil {
glog.Fatalf("Failed to initialize metadata cache store for %s: %+v", store.GetName(), err)
}
2020-09-24 10:06:44 +00:00
return filer.NewFilerStoreWrapper(store)
2020-04-29 20:26:02 +00:00
}
2020-04-30 01:20:54 +00:00
2020-09-01 07:21:19 +00:00
func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry) error {
2020-04-30 01:20:54 +00:00
mc.Lock()
defer mc.Unlock()
return mc.doInsertEntry(ctx, entry)
}
2020-09-01 07:21:19 +00:00
func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) error {
2020-09-22 23:24:13 +00:00
return mc.localStore.InsertEntry(ctx, entry)
2020-04-30 01:20:54 +00:00
}
func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
2020-04-30 01:20:54 +00:00
mc.Lock()
defer mc.Unlock()
2020-06-19 16:45:27 +00:00
oldDir, _ := oldPath.DirAndName()
if mc.visitedBoundary.HasVisited(util.FullPath(oldDir)) {
if oldPath != "" {
2020-08-17 17:07:34 +00:00
if newEntry != nil && oldPath == newEntry.FullPath {
2020-08-17 17:03:34 +00:00
// skip the unnecessary deletion
// leave the update to the following InsertEntry operation
} else {
glog.V(3).Infof("DeleteEntry %s/%s", oldPath, oldPath.Name())
2020-09-22 23:24:13 +00:00
if err := mc.localStore.DeleteEntry(ctx, oldPath); err != nil {
2020-08-17 17:03:34 +00:00
return err
}
2020-06-19 16:45:27 +00:00
}
2020-04-30 01:20:54 +00:00
}
2020-07-14 14:34:16 +00:00
} else {
2020-06-19 16:45:27 +00:00
// println("unknown old directory:", oldDir)
2020-04-30 01:20:54 +00:00
}
2020-06-19 16:45:27 +00:00
2020-04-30 01:20:54 +00:00
if newEntry != nil {
2020-06-19 16:45:27 +00:00
newDir, _ := newEntry.DirAndName()
if mc.visitedBoundary.HasVisited(util.FullPath(newDir)) {
glog.V(3).Infof("InsertEntry %s/%s", newDir, newEntry.Name())
2020-09-22 23:24:13 +00:00
if err := mc.localStore.InsertEntry(ctx, newEntry); err != nil {
2020-06-19 16:45:27 +00:00
return err
}
2020-04-30 01:20:54 +00:00
}
}
return nil
}
2020-09-01 07:21:19 +00:00
func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
2020-04-30 01:20:54 +00:00
mc.Lock()
defer mc.Unlock()
2020-09-22 23:24:13 +00:00
return mc.localStore.UpdateEntry(ctx, entry)
2020-04-30 01:20:54 +00:00
}
2020-09-01 07:21:19 +00:00
func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *filer.Entry, err error) {
2020-04-30 01:20:54 +00:00
mc.RLock()
defer mc.RUnlock()
2020-09-22 23:24:13 +00:00
entry, err = mc.localStore.FindEntry(ctx, fp)
if err != nil {
return nil, err
}
mc.mapIdFromFilerToLocal(entry)
return
2020-04-30 01:20:54 +00:00
}
func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
mc.Lock()
defer mc.Unlock()
2020-09-22 23:24:13 +00:00
return mc.localStore.DeleteEntry(ctx, fp)
2020-04-30 01:20:54 +00:00
}
2021-01-16 07:56:24 +00:00
func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error {
2020-04-30 01:20:54 +00:00
mc.RLock()
defer mc.RUnlock()
if !mc.visitedBoundary.HasVisited(dirPath) {
2021-01-16 07:56:24 +00:00
return fmt.Errorf("unsynchronized dir: %v", dirPath)
}
2021-01-16 07:56:24 +00:00
_, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) bool {
mc.mapIdFromFilerToLocal(entry)
2021-01-16 07:56:24 +00:00
return eachEntryFunc(entry)
})
if err != nil {
return err
}
2021-01-16 07:56:24 +00:00
return err
2020-04-30 01:20:54 +00:00
}
func (mc *MetaCache) Shutdown() {
mc.Lock()
defer mc.Unlock()
2020-09-22 23:24:13 +00:00
mc.localStore.Shutdown()
2020-04-30 01:20:54 +00:00
}
func (mc *MetaCache) mapIdFromFilerToLocal(entry *filer.Entry) {
entry.Attr.Uid, entry.Attr.Gid = mc.uidGidMapper.FilerToLocal(entry.Attr.Uid, entry.Attr.Gid)
}