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.

106 lines
2.6 KiB

5 years ago
  1. package meta_cache
  2. import (
  3. "context"
  4. "os"
  5. "sync"
  6. "github.com/chrislusf/seaweedfs/weed/filer2"
  7. "github.com/chrislusf/seaweedfs/weed/filer2/leveldb"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/util"
  10. "github.com/chrislusf/seaweedfs/weed/util/bounded_tree"
  11. )
  12. type MetaCache struct {
  13. actualStore filer2.FilerStore
  14. sync.RWMutex
  15. visitedBoundary *bounded_tree.BoundedTree
  16. }
  17. func NewMetaCache(dbFolder string) *MetaCache {
  18. return &MetaCache{
  19. actualStore: openMetaStore(dbFolder),
  20. visitedBoundary: bounded_tree.NewBoundedTree(),
  21. }
  22. }
  23. func openMetaStore(dbFolder string) filer2.FilerStore {
  24. os.RemoveAll(dbFolder)
  25. os.MkdirAll(dbFolder, 0755)
  26. store := &leveldb.LevelDBStore{}
  27. config := &cacheConfig{
  28. dir: dbFolder,
  29. }
  30. if err := store.Initialize(config, ""); err != nil {
  31. glog.Fatalf("Failed to initialize metadata cache store for %s: %+v", store.GetName(), err)
  32. }
  33. return store
  34. }
  35. func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer2.Entry) error {
  36. mc.Lock()
  37. defer mc.Unlock()
  38. return mc.actualStore.InsertEntry(ctx, entry)
  39. }
  40. func (mc *MetaCache) AtomicUpdateEntry(ctx context.Context, oldPath util.FullPath, newEntry *filer2.Entry) error {
  41. mc.Lock()
  42. defer mc.Unlock()
  43. oldDir, _ := oldPath.DirAndName()
  44. if mc.visitedBoundary.HasVisited(util.FullPath(oldDir)) {
  45. if oldPath != "" {
  46. if err := mc.actualStore.DeleteEntry(ctx, oldPath); err != nil {
  47. return err
  48. }
  49. }
  50. }else{
  51. // println("unknown old directory:", oldDir)
  52. }
  53. if newEntry != nil {
  54. newDir, _ := newEntry.DirAndName()
  55. if mc.visitedBoundary.HasVisited(util.FullPath(newDir)) {
  56. if err := mc.actualStore.InsertEntry(ctx, newEntry); err != nil {
  57. return err
  58. }
  59. }
  60. }
  61. return nil
  62. }
  63. func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer2.Entry) error {
  64. mc.Lock()
  65. defer mc.Unlock()
  66. return mc.actualStore.UpdateEntry(ctx, entry)
  67. }
  68. func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *filer2.Entry, err error) {
  69. mc.RLock()
  70. defer mc.RUnlock()
  71. return mc.actualStore.FindEntry(ctx, fp)
  72. }
  73. func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
  74. mc.Lock()
  75. defer mc.Unlock()
  76. return mc.actualStore.DeleteEntry(ctx, fp)
  77. }
  78. func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int) ([]*filer2.Entry, error) {
  79. mc.RLock()
  80. defer mc.RUnlock()
  81. return mc.actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
  82. }
  83. func (mc *MetaCache) Shutdown() {
  84. mc.Lock()
  85. defer mc.Unlock()
  86. mc.actualStore.Shutdown()
  87. }