diff --git a/.travis.yml b/.travis.yml index d2bf24341..48ccadec2 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,6 +1,5 @@ language: go go: -- 1.8 - 1.9 os: - linux @@ -17,4 +16,4 @@ deploy: repo: kopia/kopia go: 1.6 git: - depth: 9999999 \ No newline at end of file + depth: 9999999 diff --git a/cli/command_metadata_ls.go b/cli/command_metadata_ls.go index 5ab72ab42..2153b830b 100644 --- a/cli/command_metadata_ls.go +++ b/cli/command_metadata_ls.go @@ -18,7 +18,7 @@ func init() { func listMetadataItems(context *kingpin.ParseContext) error { rep := mustOpenRepository(nil) - entries, err := rep.ListMetadata(*metadataListPrefix, -1) + entries, err := rep.ListMetadata(*metadataListPrefix) if err != nil { return err } diff --git a/cli/command_object_cleanup.go b/cli/command_object_cleanup.go index b7f8430aa..03d8700d0 100644 --- a/cli/command_object_cleanup.go +++ b/cli/command_object_cleanup.go @@ -155,7 +155,7 @@ func runCleanupCommand(context *kingpin.ParseContext) error { mgr := snapshot.NewManager(rep) log.Printf("Listing active snapshots...") - snapshotNames, err := mgr.ListSnapshotManifests(nil, -1) + snapshotNames, err := mgr.ListSnapshotManifests(nil) if err != nil { return err } diff --git a/cli/command_repository_migrate.go b/cli/command_repository_migrate.go index aa680ba52..24e33eed7 100644 --- a/cli/command_repository_migrate.go +++ b/cli/command_repository_migrate.go @@ -48,7 +48,7 @@ func runMigrateCommand(context *kingpin.ParseContext) error { log.Printf("migrating source %v", s) - manifests, err := sourceSM.ListSnapshotManifests(s, -1) + manifests, err := sourceSM.ListSnapshotManifests(s) if err != nil { return fmt.Errorf("unable to list snapshot manifests for %v: %v", s, err) } diff --git a/cli/command_snapshot_create.go b/cli/command_snapshot_create.go index 07f6877a8..90b3cc617 100644 --- a/cli/command_snapshot_create.go +++ b/cli/command_snapshot_create.go @@ -83,7 +83,7 @@ func runBackupCommand(c *kingpin.ParseContext) error { return fmt.Errorf("description too long") } - previous, err := mgr.ListSnapshots(sourceInfo, 1) + previous, err := mgr.ListSnapshots(sourceInfo) if err != nil { return fmt.Errorf("error listing previous backups: %v", err) } diff --git a/cli/command_snapshot_expire.go b/cli/command_snapshot_expire.go index c664b061b..4f34ddf07 100644 --- a/cli/command_snapshot_expire.go +++ b/cli/command_snapshot_expire.go @@ -110,7 +110,7 @@ func getSnapshotNamesToExpire(mgr *snapshot.Manager) ([]string, error) { if *snapshotExpireAll { fmt.Fprintf(os.Stderr, "Scanning all active snapshots...\n") - return mgr.ListSnapshotManifests(nil, -1) + return mgr.ListSnapshotManifests(nil) } var result []string @@ -123,7 +123,7 @@ func getSnapshotNamesToExpire(mgr *snapshot.Manager) ([]string, error) { log.Printf("Looking for snapshots of %v", src) - matches, err := mgr.ListSnapshotManifests(&src, -1) + matches, err := mgr.ListSnapshotManifests(&src) if err != nil { return nil, fmt.Errorf("error listing snapshots for %v: %v", src, err) } diff --git a/cli/command_snapshot_list.go b/cli/command_snapshot_list.go index c13029c33..7b7b5086b 100644 --- a/cli/command_snapshot_list.go +++ b/cli/command_snapshot_list.go @@ -24,7 +24,7 @@ var ( func findBackups(mgr *snapshot.Manager, sourceInfo snapshot.SourceInfo) (manifestIDs []string, relPath string, err error) { for len(sourceInfo.Path) > 0 { - list, err := mgr.ListSnapshotManifests(&sourceInfo, -1) + list, err := mgr.ListSnapshotManifests(&sourceInfo) if err != nil { return nil, "", err } @@ -72,7 +72,7 @@ func runBackupsCommand(context *kingpin.ParseContext) error { relPath = "/" + relPath } } else { - previous, err = mgr.ListSnapshotManifests(nil, -1) + previous, err = mgr.ListSnapshotManifests(nil) } if err != nil { diff --git a/fs/repofs/source_snapshots.go b/fs/repofs/source_snapshots.go index 68ecf01b7..df6769eb4 100644 --- a/fs/repofs/source_snapshots.go +++ b/fs/repofs/source_snapshots.go @@ -35,7 +35,7 @@ func safeName(path string) string { } func (s *sourceSnapshots) Readdir() (fs.Entries, error) { - manifests, err := s.snapshotManager.ListSnapshots(s.src, -1) + manifests, err := s.snapshotManager.ListSnapshots(s.src) if err != nil { return nil, err } diff --git a/repo/initialize.go b/repo/initialize.go index e638245eb..bd02d5b7e 100644 --- a/repo/initialize.go +++ b/repo/initialize.go @@ -39,12 +39,17 @@ func Initialize(st blob.Storage, opt *NewRepositoryOptions, creds auth.Credentia opt = &NewRepositoryOptions{} } + cache, err := newMetadataCache(st) + if err != nil { + return err + } + mm := MetadataManager{ storage: st, + cache: cache, format: metadataFormatFromOptions(opt), } - var err error mm.masterKey, err = creds.GetMasterKey(mm.format.SecurityOptions) if err != nil { return err diff --git a/repo/metadata_cache.go b/repo/metadata_cache.go new file mode 100644 index 000000000..94b7466d2 --- /dev/null +++ b/repo/metadata_cache.go @@ -0,0 +1,157 @@ +package repo + +import ( + "fmt" + "sort" + "strings" + "sync" + + "github.com/kopia/kopia/blob" +) + +const ( + // MetadataBlockPrefix is a prefix used for metadata blocks in repository storage. + MetadataBlockPrefix = "VLT" +) + +type metadataCache struct { + st blob.Storage + + mu sync.Mutex + sortedNames []string + nameToCacheID map[string]string + cachedData map[string][]byte +} + +func (mc *metadataCache) ListBlocks(prefix string) ([]string, error) { + var result []string + + mc.mu.Lock() + p := sort.SearchStrings(mc.sortedNames, prefix) + for p < len(mc.sortedNames) && strings.HasPrefix(mc.sortedNames[p], prefix) { + if !isReservedName(mc.sortedNames[p]) { + result = append(result, mc.sortedNames[p]) + } + p++ + } + mc.mu.Unlock() + return result, nil +} + +func (mc *metadataCache) GetBlock(name string) ([]byte, error) { + mc.mu.Lock() + cid := mc.nameToCacheID[name] + if cid == "" { + mc.mu.Unlock() + return nil, blob.ErrBlockNotFound + } + + // see if the data is cached + if data, ok := mc.cachedData[cid]; ok { + mc.mu.Unlock() + return cloneBytes(data), nil + } + + mc.mu.Unlock() + + // not cached, fetch from the storage + b, err := mc.st.GetBlock(MetadataBlockPrefix+name, 0, -1) + if err != nil { + return nil, err + } + + // now race to add to cache, does not matter who wins. + mc.mu.Lock() + mc.cachedData[cid] = b + mc.mu.Unlock() + return cloneBytes(b), nil +} + +func (mc *metadataCache) PutBlock(name string, data []byte) error { + if err := mc.st.PutBlock(MetadataBlockPrefix+name, data); err != nil { + return err + } + + mc.mu.Lock() + cid := fmt.Sprintf("%v-new-%x", name, randomBytes(8)) + mc.nameToCacheID[name] = cid + p := sort.SearchStrings(mc.sortedNames, name) + if p >= len(mc.sortedNames) || mc.sortedNames[p] != name { + // Name not present, p is the index where p should be inserted. + mc.sortedNames = append(append( + append([]string(nil), mc.sortedNames[0:p]...), + name), + mc.sortedNames[p:]...) + } + + mc.mu.Unlock() + + return nil +} + +func (mc *metadataCache) DeleteBlock(name string) error { + mc.mu.Lock() + if cid := mc.nameToCacheID[name]; cid != "" { + delete(mc.nameToCacheID, name) + delete(mc.cachedData, cid) + + // Delete from sortedNames + p := sort.SearchStrings(mc.sortedNames, name) + if p < len(mc.sortedNames) && mc.sortedNames[p] == name { + // Found at index 'p' build a new slice from [0..p), [p+1,...) + newSlice := mc.sortedNames[0:p] + for _, n := range mc.sortedNames[p+1:] { + newSlice = append(newSlice, n) + } + mc.sortedNames = newSlice + } + } + mc.mu.Unlock() + + return mc.st.DeleteBlock(MetadataBlockPrefix + name) +} + +// Refresh refreshes the list of blocks in the cache, but does not load or expire previously cached. +func (mc *metadataCache) Refresh() error { + var sortedNames []string + nameToCacheID := map[string]string{} + + ch, cancel := mc.st.ListBlocks(MetadataBlockPrefix) + defer cancel() + for it := range ch { + if it.Error != nil { + return it.Error + } + + n := strings.TrimPrefix(it.BlockID, MetadataBlockPrefix) + sortedNames = append(sortedNames, n) + nameToCacheID[n] = fmt.Sprintf("%v-%v-%v", it.BlockID, it.Length, it.TimeStamp.UnixNano()) + } + + mc.setLoaded(sortedNames, nameToCacheID) + return nil +} + +func (mc *metadataCache) setLoaded(sortedNames []string, nameToCacheID map[string]string) { + mc.mu.Lock() + mc.sortedNames = sortedNames + mc.nameToCacheID = nameToCacheID + mc.mu.Unlock() +} + +func cloneBytes(d []byte) []byte { + return append([]byte(nil), d...) +} + +func newMetadataCache(st blob.Storage) (*metadataCache, error) { + c := &metadataCache{ + st: st, + nameToCacheID: make(map[string]string), + cachedData: make(map[string][]byte), + } + if err := c.Refresh(); err != nil { + return nil, err + } + + return c, nil +} diff --git a/repo/metadata_manager.go b/repo/metadata_manager.go index 0ace0c7e7..c4c7e0670 100644 --- a/repo/metadata_manager.go +++ b/repo/metadata_manager.go @@ -9,7 +9,6 @@ import ( "errors" "fmt" "io" - "strings" "sync" "github.com/kopia/kopia/auth" @@ -25,9 +24,6 @@ const ( ) const ( - // MetadataBlockPrefix is a prefix used for metadata blocks in repository storage. - MetadataBlockPrefix = "VLT" - parallelFetches = 5 ) @@ -59,6 +55,7 @@ func init() { // in a repository. type MetadataManager struct { storage blob.Storage + cache *metadataCache format config.MetadataFormat repoConfig config.EncryptedRepositoryConfig @@ -94,11 +91,11 @@ func (mm *MetadataManager) writeEncryptedBlock(itemID string, content []byte) er content = nonce[0 : nonceLength+len(b)] } - return mm.storage.PutBlock(MetadataBlockPrefix+itemID, content) + return mm.cache.PutBlock(itemID, content) } func (mm *MetadataManager) readEncryptedBlock(itemID string) ([]byte, error) { - content, err := mm.storage.GetBlock(MetadataBlockPrefix+itemID, 0, -1) + content, err := mm.cache.GetBlock(itemID) if err != nil { if err == blob.ErrBlockNotFound { return nil, ErrMetadataNotFound @@ -203,34 +200,13 @@ func (mm *MetadataManager) putJSON(id string, content interface{}) error { } // ListMetadata returns the list of metadata items matching the specified prefix. -// The 'limit' parameter specifies the maximum number of items to retrieve (-1 == unlimited). -func (mm *MetadataManager) ListMetadata(prefix string, limit int) ([]string, error) { - var result []string - - ch, cancel := mm.storage.ListBlocks(MetadataBlockPrefix + prefix) - defer cancel() - for b := range ch { - if limit == 0 { - break - } - if b.Error != nil { - return result, b.Error - } - - itemID := strings.TrimPrefix(b.BlockID, MetadataBlockPrefix) - if !isReservedName(itemID) { - result = append(result, itemID) - } - if limit > 0 { - limit-- - } - } - return result, nil +func (mm *MetadataManager) ListMetadata(prefix string) ([]string, error) { + return mm.cache.ListBlocks(prefix) } -// ListMetadataContents retrieves metadata contents for all items starting with a given prefix and up to a specified limit. -func (mm *MetadataManager) ListMetadataContents(prefix string, limit int) (map[string][]byte, error) { - itemIDs, err := mm.ListMetadata(prefix, limit) +// ListMetadataContents retrieves metadata contents for all items starting with a given prefix. +func (mm *MetadataManager) ListMetadataContents(prefix string) (map[string][]byte, error) { + itemIDs, err := mm.ListMetadata(prefix) if err != nil { return nil, err } @@ -260,7 +236,7 @@ func (mm *MetadataManager) RemoveMetadata(itemID string) error { return err } - return mm.storage.DeleteBlock(MetadataBlockPrefix + itemID) + return mm.cache.DeleteBlock(itemID) } // RemoveMany efficiently removes multiple metadata items in parallel. @@ -295,8 +271,14 @@ func (mm *MetadataManager) RemoveMany(itemIDs []string) error { // newMetadataManager opens a MetadataManager for given storage and credentials. func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManager, error) { + cache, err := newMetadataCache(st) + if err != nil { + return nil, err + } + mm := MetadataManager{ storage: st, + cache: cache, } var wg sync.WaitGroup @@ -304,13 +286,13 @@ func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManag var blocks [4][]byte f := func(index int, name string) { - blocks[index], _ = st.GetBlock(name, 0, -1) + blocks[index], _ = mm.cache.GetBlock(name) wg.Done() } wg.Add(2) - go f(0, MetadataBlockPrefix+formatBlockID) - go f(1, MetadataBlockPrefix+repositoryConfigBlockID) + go f(0, formatBlockID) + go f(1, repositoryConfigBlockID) wg.Wait() if blocks[0] == nil { @@ -318,7 +300,7 @@ func newMetadataManager(st blob.Storage, creds auth.Credentials) (*MetadataManag } var offset = 0 - err := json.Unmarshal(blocks[offset], &mm.format) + err = json.Unmarshal(blocks[offset], &mm.format) if err != nil { return nil, err } diff --git a/repo/metadata_manager_test.go b/repo/metadata_manager_test.go index 66708f92b..f7da3e8ba 100644 --- a/repo/metadata_manager_test.go +++ b/repo/metadata_manager_test.go @@ -41,30 +41,19 @@ func TestMetadataManager(t *testing.T) { return } - v1, err := newMetadataManager(st, creds) + v, err := newMetadataManager(st, creds) if err != nil { t.Errorf("can't open first metadata manager: %v", err) return } - v2, err := newMetadataManager(st, creds) - if err != nil { - t.Errorf("can't open second metadata manager: %v", err) - return - } - - cfg, err := v1.connectionConfiguration() + cfg, err := v.connectionConfiguration() if err != nil { t.Errorf("error getting token1 %v", err) } - cfg2, err := v2.connectionConfiguration() - if err != nil { - t.Errorf("error getting token2 %v", err) - } - - if !reflect.DeepEqual(cfg, cfg2) { - t.Errorf("configurations are different: %+v vs %+v", cfg, cfg2) + if got, want := cfg.ConnectionInfo.Type, "filesystem"; got != want { + t.Errorf("invalid connection type: %q, want %q", got, want) } _, err = newMetadataManager(st, otherCreds) @@ -73,54 +62,72 @@ func TestMetadataManager(t *testing.T) { return } - if err := v1.PutMetadata("foo", []byte("test1")); err != nil { + if err := v.PutMetadata("foo", []byte("test1")); err != nil { t.Errorf("error putting: %v", err) } - if err := v2.PutMetadata("bar", []byte("test2")); err != nil { + if err := v.PutMetadata("bar", []byte("test2")); err != nil { t.Errorf("error putting: %v", err) } - if err := v1.PutMetadata("baz", []byte("test3")); err != nil { + if err := v.PutMetadata("baz", []byte("test3")); err != nil { t.Errorf("error putting: %v", err) } - // Verify contents of metadata items for both managers. - for _, v := range []*MetadataManager{v1, v2} { - assertMetadataItem(t, v, "foo", "test1") - assertMetadataItem(t, v, "bar", "test2") - assertMetadataItem(t, v, "baz", "test3") - assertMetadataItemNotFound(t, v, "x") + assertMetadataItem(t, v, "foo", "test1") + assertMetadataItem(t, v, "bar", "test2") + assertMetadataItem(t, v, "baz", "test3") + assertMetadataItemNotFound(t, v, "x") - assertMetadataItems(t, v, "x", nil) - assertMetadataItems(t, v, "f", []string{"foo"}) - assertMetadataItems(t, v, "ba", []string{"bar", "baz"}) - assertMetadataItems(t, v, "be", nil) - assertMetadataItems(t, v, "baz", []string{"baz"}) - assertMetadataItems(t, v, "bazx", nil) + assertMetadataItems(t, v, "x", nil) + assertMetadataItems(t, v, "f", []string{"foo"}) + assertMetadataItems(t, v, "ba", []string{"bar", "baz"}) + assertMetadataItems(t, v, "be", nil) + assertMetadataItems(t, v, "baz", []string{"baz"}) + assertMetadataItems(t, v, "bazx", nil) - assertReservedName(t, v, formatBlockID) - assertReservedName(t, v, repositoryConfigBlockID) + assertReservedName(t, v, formatBlockID) + assertReservedName(t, v, repositoryConfigBlockID) + + v.RemoveMetadata("bar") + + assertMetadataItem(t, v, "foo", "test1") + assertMetadataItemNotFound(t, v, "bar") + assertMetadataItem(t, v, "baz", "test3") + + assertMetadataItems(t, v, "x", nil) + assertMetadataItems(t, v, "f", []string{"foo"}) + assertMetadataItems(t, v, "ba", []string{"baz"}) + assertMetadataItems(t, v, "be", nil) + assertMetadataItems(t, v, "baz", []string{"baz"}) + assertMetadataItems(t, v, "bazx", nil) + + v.RemoveMetadata("baz") + assertMetadataItemNotFound(t, v, "baz") + v.RemoveMetadata("baz") + assertMetadataItemNotFound(t, v, "baz") + + assertMetadataItem(t, v, "foo", "test1") + if err := v.PutMetadata("baz", []byte("test4")); err != nil { + t.Errorf("error putting: %v", err) + } + assertMetadataItem(t, v, "baz", "test4") + + v2, err := newMetadataManager(st, creds) + if err != nil { + t.Errorf("can't open first metadata manager: %v", err) + return } - v1.RemoveMetadata("bar") - - for _, v := range []*MetadataManager{v1, v2} { - assertMetadataItem(t, v, "foo", "test1") - assertMetadataItemNotFound(t, v, "bar") - assertMetadataItem(t, v, "baz", "test3") - - assertMetadataItems(t, v, "x", nil) - assertMetadataItems(t, v, "f", []string{"foo"}) - assertMetadataItems(t, v, "ba", []string{"baz"}) - assertMetadataItems(t, v, "be", nil) - assertMetadataItems(t, v, "baz", []string{"baz"}) - assertMetadataItems(t, v, "bazx", nil) - } + assertMetadataItem(t, v2, "foo", "test1") + assertMetadataItemNotFound(t, v2, "bar") + assertMetadataItem(t, v2, "baz", "test4") } func assertMetadataItem(t *testing.T, v *MetadataManager, itemID string, expectedData string) { + t.Helper() b, err := v.GetMetadata(itemID) if err != nil { t.Errorf("error getting item %v: %v", itemID, err) + return } bs := string(b) @@ -155,7 +162,8 @@ func assertReservedNameError(t *testing.T, method string, itemID string, err err } func assertMetadataItems(t *testing.T, v *MetadataManager, prefix string, expected []string) { - res, err := v.ListMetadata(prefix, -1) + t.Helper() + res, err := v.ListMetadata(prefix) if err != nil { t.Errorf("error listing items beginning with %v: %v", prefix, err) } diff --git a/repo/pack_manager.go b/repo/pack_manager.go index 8ce79c13e..a4a8b3bf0 100644 --- a/repo/pack_manager.go +++ b/repo/pack_manager.go @@ -208,7 +208,7 @@ func (p *packManager) ensurePackIndexesLoaded() (map[string]*packIndex, error) { p.mu.Lock() defer p.mu.Unlock() - m, err := p.metadataManager.ListMetadataContents(packIDPrefix, -1) + m, err := p.metadataManager.ListMetadataContents(packIDPrefix) if err != nil { return nil, err } diff --git a/snapshot/manager.go b/snapshot/manager.go index 816bc9fdb..f58149748 100644 --- a/snapshot/manager.go +++ b/snapshot/manager.go @@ -33,7 +33,7 @@ type Manager struct { // ListSources lists all snapshot sources. func (m *Manager) ListSources() ([]*SourceInfo, error) { - names, err := m.repository.ListMetadata(snapshotPrefix, -1) + names, err := m.repository.ListMetadata(snapshotPrefix) if err != nil { return nil, err } @@ -63,8 +63,8 @@ func (m *Manager) ListSources() ([]*SourceInfo, error) { } // ListSnapshots lists all snapshots for a given source. -func (m *Manager) ListSnapshots(si *SourceInfo, limit int) ([]*Manifest, error) { - names, err := m.repository.ListMetadata(m.snapshotIDPrefix(si), limit) +func (m *Manager) ListSnapshots(si *SourceInfo) ([]*Manifest, error) { + names, err := m.repository.ListMetadata(m.snapshotIDPrefix(si)) if err != nil { return nil, err } @@ -141,14 +141,13 @@ func (m *Manager) LoadSnapshots(names []string) ([]*Manifest, error) { } // ListSnapshotManifests returns the list of snapshot manifests for a given source or all sources if nil. -// Limit specifies how mamy manifests to retrieve (-1 == unlimited). -func (m *Manager) ListSnapshotManifests(src *SourceInfo, limit int) ([]string, error) { +func (m *Manager) ListSnapshotManifests(src *SourceInfo) ([]string, error) { prefix := snapshotPrefix if src != nil { prefix = m.snapshotIDPrefix(src) } - return m.repository.ListMetadata(prefix, limit) + return m.repository.ListMetadata(prefix) } // GetEffectivePolicy calculates effective snapshot policy for a given source by combining the source-specifc policy (if any) @@ -271,7 +270,7 @@ func (m *Manager) getPolicyItem(itemID string) (*Policy, error) { // ListPolicies returns a list of all snapshot policies. func (m *Manager) ListPolicies() ([]*Policy, error) { - names, err := m.repository.ListMetadata(policyPrefix, -1) + names, err := m.repository.ListMetadata(policyPrefix) if err != nil { return nil, err }