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.

134 lines
3.4 KiB

5 years ago
4 years ago
5 years ago
  1. package meta_cache
  2. import (
  3. "context"
  4. "os"
  5. "sync"
  6. "github.com/chrislusf/seaweedfs/weed/filer"
  7. "github.com/chrislusf/seaweedfs/weed/filer/leveldb"
  8. "github.com/chrislusf/seaweedfs/weed/glog"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. "github.com/chrislusf/seaweedfs/weed/util/bounded_tree"
  12. )
  13. // need to have logic similar to FilerStoreWrapper
  14. // e.g. fill fileId field for chunks
  15. type MetaCache struct {
  16. actualStore filer.FilerStore
  17. sync.RWMutex
  18. visitedBoundary *bounded_tree.BoundedTree
  19. }
  20. func NewMetaCache(dbFolder string) *MetaCache {
  21. return &MetaCache{
  22. actualStore: openMetaStore(dbFolder),
  23. visitedBoundary: bounded_tree.NewBoundedTree(),
  24. }
  25. }
  26. func openMetaStore(dbFolder string) filer.FilerStore {
  27. os.RemoveAll(dbFolder)
  28. os.MkdirAll(dbFolder, 0755)
  29. store := &leveldb.LevelDBStore{}
  30. config := &cacheConfig{
  31. dir: dbFolder,
  32. }
  33. if err := store.Initialize(config, ""); err != nil {
  34. glog.Fatalf("Failed to initialize metadata cache store for %s: %+v", store.GetName(), err)
  35. }
  36. return store
  37. }
  38. func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry) error {
  39. mc.Lock()
  40. defer mc.Unlock()
  41. return mc.doInsertEntry(ctx, entry)
  42. }
  43. func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) error {
  44. filer_pb.BeforeEntrySerialization(entry.Chunks)
  45. return mc.actualStore.InsertEntry(ctx, entry)
  46. }
  47. func (mc *MetaCache) AtomicUpdateEntry(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
  48. mc.Lock()
  49. defer mc.Unlock()
  50. oldDir, _ := oldPath.DirAndName()
  51. if mc.visitedBoundary.HasVisited(util.FullPath(oldDir)) {
  52. if oldPath != "" {
  53. if newEntry != nil && oldPath == newEntry.FullPath {
  54. // skip the unnecessary deletion
  55. // leave the update to the following InsertEntry operation
  56. } else {
  57. if err := mc.actualStore.DeleteEntry(ctx, oldPath); err != nil {
  58. return err
  59. }
  60. }
  61. }
  62. } else {
  63. // println("unknown old directory:", oldDir)
  64. }
  65. if newEntry != nil {
  66. newDir, _ := newEntry.DirAndName()
  67. if mc.visitedBoundary.HasVisited(util.FullPath(newDir)) {
  68. if err := mc.actualStore.InsertEntry(ctx, newEntry); err != nil {
  69. return err
  70. }
  71. }
  72. }
  73. return nil
  74. }
  75. func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
  76. mc.Lock()
  77. defer mc.Unlock()
  78. filer_pb.BeforeEntrySerialization(entry.Chunks)
  79. return mc.actualStore.UpdateEntry(ctx, entry)
  80. }
  81. func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *filer.Entry, err error) {
  82. mc.RLock()
  83. defer mc.RUnlock()
  84. entry, err = mc.actualStore.FindEntry(ctx, fp)
  85. if err != nil {
  86. return nil, err
  87. }
  88. filer_pb.AfterEntryDeserialization(entry.Chunks)
  89. return
  90. }
  91. func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
  92. mc.Lock()
  93. defer mc.Unlock()
  94. return mc.actualStore.DeleteEntry(ctx, fp)
  95. }
  96. func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int) ([]*filer.Entry, error) {
  97. mc.RLock()
  98. defer mc.RUnlock()
  99. entries, err := mc.actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
  100. if err != nil {
  101. return nil, err
  102. }
  103. for _, entry := range entries {
  104. filer_pb.AfterEntryDeserialization(entry.Chunks)
  105. }
  106. return entries, err
  107. }
  108. func (mc *MetaCache) Shutdown() {
  109. mc.Lock()
  110. defer mc.Unlock()
  111. mc.actualStore.Shutdown()
  112. }