From c4ab086d5fc0e9c61da167a111222c166c203626 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Tue, 15 Mar 2022 23:57:04 -0700 Subject: [PATCH] chore(repository): streamlined internal content cache API (#1828) moved from ./repo/content to ./internal/cache --- cli/command_cache_sync.go | 42 ++++++- internal/cache/content_cache.go | 23 ++++ internal/cache/content_cache_data.go | 94 ++++++++++++++ .../cache}/content_cache_data_test.go | 22 ++-- .../cache}/content_cache_metadata.go | 57 +++------ internal/cache/content_cache_metadata_test.go | 81 ++++++++++++ internal/cache/content_cache_passthrough.go | 32 +++++ .../cache}/content_cache_test.go | 61 ++++----- repo/content/committed_read_manager.go | 14 +-- repo/content/content_cache.go | 14 --- repo/content/content_cache_data.go | 88 ------------- repo/content/content_cache_metadata_test.go | 118 ------------------ repo/content/content_cache_passthrough.go | 24 ---- repo/content/content_manager.go | 13 +- repo/content/content_manager_lock_free.go | 2 +- repo/content/content_manager_test.go | 34 ++--- repo/content/content_prefetch.go | 4 +- 17 files changed, 355 insertions(+), 368 deletions(-) create mode 100644 internal/cache/content_cache.go create mode 100644 internal/cache/content_cache_data.go rename {repo/content => internal/cache}/content_cache_data_test.go (79%) rename {repo/content => internal/cache}/content_cache_metadata.go (61%) create mode 100644 internal/cache/content_cache_metadata_test.go create mode 100644 internal/cache/content_cache_passthrough.go rename {repo/content => internal/cache}/content_cache_test.go (82%) delete mode 100644 repo/content/content_cache.go delete mode 100644 repo/content/content_cache_data.go delete mode 100644 repo/content/content_cache_metadata_test.go delete mode 100644 repo/content/content_cache_passthrough.go diff --git a/cli/command_cache_sync.go b/cli/command_cache_sync.go index 110bb43e6..f5cfd1d4a 100644 --- a/cli/command_cache_sync.go +++ b/cli/command_cache_sync.go @@ -3,17 +3,53 @@ package cli import ( "context" + "github.com/pkg/errors" + "golang.org/x/sync/errgroup" + "github.com/kopia/kopia/repo" + "github.com/kopia/kopia/repo/blob" + "github.com/kopia/kopia/repo/content" ) -type commandCacheSync struct{} +type commandCacheSync struct { + parallel int +} func (c *commandCacheSync) setup(svc appServices, parent commandParent) { cmd := parent.Command("sync", "Synchronizes the metadata cache with blobs in storage") + cmd.Flag("parallel", "Fetch parallelism").Default("16").IntVar(&c.parallel) cmd.Action(svc.directRepositoryWriteAction(c.run)) } func (c *commandCacheSync) run(ctx context.Context, rep repo.DirectRepositoryWriter) error { - // nolint:wrapcheck - return rep.ContentManager().SyncMetadataCache(ctx) + eg, ctx := errgroup.WithContext(ctx) + + ch := make(chan blob.ID, c.parallel) + + // workers that will prefetch blobs. + for i := 0; i < c.parallel; i++ { + eg.Go(func() error { + for blobID := range ch { + if err := rep.ContentManager().MetadataCache().PrefetchBlob(ctx, blobID); err != nil { + return errors.Wrap(err, "error prefetching blob") + } + } + + return nil + }) + } + + // populate channel with blob IDs. + eg.Go(func() error { + defer close(ch) + + // nolint:wrapcheck + return rep.BlobReader().ListBlobs(ctx, content.PackBlobIDPrefixSpecial, func(bm blob.Metadata) error { + ch <- bm.BlobID + + return nil + }) + }) + + return errors.Wrap(eg.Wait(), "error synchronizing metadata cache") } diff --git a/internal/cache/content_cache.go b/internal/cache/content_cache.go new file mode 100644 index 000000000..10a04e014 --- /dev/null +++ b/internal/cache/content_cache.go @@ -0,0 +1,23 @@ +package cache + +import ( + "context" + + "github.com/kopia/kopia/internal/gather" + "github.com/kopia/kopia/repo/blob" +) + +// ContentCache caches contents stored in pack blobs. +type ContentCache interface { + Close(ctx context.Context) + GetContent(ctx context.Context, contentID string, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error + PrefetchBlob(ctx context.Context, blobID blob.ID) error + CacheStorage() Storage +} + +// SyncableContentCache caches contents stored in pack blobs and supports synchronizing. +type SyncableContentCache interface { + ContentCache + + Sync(ctx context.Context, blobPrefix blob.ID) error +} diff --git a/internal/cache/content_cache_data.go b/internal/cache/content_cache_data.go new file mode 100644 index 000000000..d41b6dbab --- /dev/null +++ b/internal/cache/content_cache_data.go @@ -0,0 +1,94 @@ +package cache + +import ( + "context" + + "github.com/pkg/errors" + "go.opencensus.io/stats" + + "github.com/kopia/kopia/internal/gather" + "github.com/kopia/kopia/repo/blob" +) + +type contentCacheForData struct { + pc *PersistentCache + st blob.Storage +} + +// ContentIDCacheKey computes the cache key for the provided content ID. +func ContentIDCacheKey(contentID string) string { + // content IDs with odd length have a single-byte prefix. + // move the prefix to the end of cache key to make sure the top level shard is spread 256 ways. + if len(contentID)%2 == 1 { + return contentID[1:] + contentID[0:1] + } + + return contentID +} + +// BlobIDCacheKey computes the cache key for the provided blob ID. +func BlobIDCacheKey(id blob.ID) string { + return string(id[1:] + id[0:1]) +} + +func (c *contentCacheForData) GetContent(ctx context.Context, contentID string, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error { + if c.pc.GetPartial(ctx, BlobIDCacheKey(blobID), offset, length, output) { + return nil + } + + output.Reset() + + // nolint:wrapcheck + return c.pc.GetOrLoad(ctx, ContentIDCacheKey(contentID), func(output *gather.WriteBuffer) error { + // nolint:wrapcheck + return c.st.GetBlob(ctx, blobID, offset, length, output) + }, output) +} + +func (c *contentCacheForData) Close(ctx context.Context) { + c.pc.Close(ctx) +} + +func (c contentCacheForData) PrefetchBlob(ctx context.Context, blobID blob.ID) error { + var blobData gather.WriteBuffer + defer blobData.Close() + + if c.pc.GetPartial(ctx, BlobIDCacheKey(blobID), 0, 1, &blobData) { + return nil + } + + // read the entire blob + if err := c.st.GetBlob(ctx, blobID, 0, -1, &blobData); err != nil { + stats.Record(ctx, MetricMissErrors.M(1)) + // nolint:wrapcheck + return err + } + + stats.Record(ctx, MetricMissBytes.M(int64(blobData.Length()))) + + // store the whole blob in the + c.pc.Put(ctx, BlobIDCacheKey(blobID), blobData.Bytes()) + + return nil +} + +func (c contentCacheForData) CacheStorage() Storage { + return c.pc.cacheStorage +} + +// NewContentCacheForData creates new content cache for data contents. +func NewContentCacheForData(ctx context.Context, st blob.Storage, cacheStorage Storage, sweep SweepSettings, hmacSecret []byte) (ContentCache, error) { + if cacheStorage == nil { + return passthroughContentCache{st}, nil + } + + pc, err := NewPersistentCache(ctx, "content cache", cacheStorage, ChecksumProtection(hmacSecret), sweep) + if err != nil { + return nil, errors.Wrap(err, "unable to create base cache") + } + + return &contentCacheForData{ + st: st, + pc: pc, + }, nil +} diff --git a/repo/content/content_cache_data_test.go b/internal/cache/content_cache_data_test.go similarity index 79% rename from repo/content/content_cache_data_test.go rename to internal/cache/content_cache_data_test.go index 58a5f3fd9..835ff2892 100644 --- a/repo/content/content_cache_data_test.go +++ b/internal/cache/content_cache_data_test.go @@ -1,4 +1,4 @@ -package content +package cache_test import ( "testing" @@ -21,7 +21,7 @@ func TestContentCacheForData(t *testing.T) { cacheData := blobtesting.DataMap{} metadataCacheStorage := blobtesting.NewMapStorage(cacheData, nil, nil).(cache.Storage) - dataCache, err := newContentCacheForData(ctx, underlying, metadataCacheStorage, cache.SweepSettings{ + dataCache, err := cache.NewContentCacheForData(ctx, underlying, metadataCacheStorage, cache.SweepSettings{ MaxSizeBytes: 100, }, []byte{1, 2, 3, 4}) require.NoError(t, err) @@ -30,14 +30,14 @@ func TestContentCacheForData(t *testing.T) { defer tmp.Close() // get something we don't have in the underlying storage - require.ErrorIs(t, dataCache.getContent(ctx, "key1", "blob1", 0, 3, &tmp), blob.ErrBlobNotFound) + require.ErrorIs(t, dataCache.GetContent(ctx, "key1", "blob1", 0, 3, &tmp), blob.ErrBlobNotFound) require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, dataCache.getContent(ctx, "xkey1", "blob1", 0, 3, &tmp)) + require.NoError(t, dataCache.GetContent(ctx, "xkey1", "blob1", 0, 3, &tmp)) require.Equal(t, []byte{1, 2, 3}, tmp.ToByteSlice()) - require.NoError(t, dataCache.getContent(ctx, "xkey2", "blob1", 3, 3, &tmp)) + require.NoError(t, dataCache.GetContent(ctx, "xkey2", "blob1", 3, 3, &tmp)) require.Equal(t, []byte{4, 5, 6}, tmp.ToByteSlice()) // cache has 2 entries, one for each section of the blob. @@ -47,11 +47,11 @@ func TestContentCacheForData(t *testing.T) { require.Contains(t, cacheData, blob.ID("key1x")) require.Contains(t, cacheData, blob.ID("key2x")) - dataCache.close(ctx) + dataCache.Close(ctx) // get slice with cache miss require.NoError(t, underlying.PutBlob(ctx, "blob2", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, dataCache.getContent(ctx, "aaa1", "blob2", 3, 3, &tmp)) + require.NoError(t, dataCache.GetContent(ctx, "aaa1", "blob2", 3, 3, &tmp)) require.Equal(t, []byte{4, 5, 6}, tmp.ToByteSlice()) // even-length content IDs are not mangled @@ -59,12 +59,12 @@ func TestContentCacheForData(t *testing.T) { require.Contains(t, cacheData, blob.ID("aaa1")) require.NoError(t, underlying.PutBlob(ctx, "pblob3", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - dataCache.prefetchBlob(ctx, "pblob3") + dataCache.PrefetchBlob(ctx, "pblob3") require.NoError(t, underlying.DeleteBlob(ctx, "pblob3")) // make sure blob ID is properly mangled require.Contains(t, cacheData, blob.ID("blob3p")) - require.NoError(t, dataCache.getContent(ctx, "aaa3", "pblob3", 2, 4, &tmp)) + require.NoError(t, dataCache.GetContent(ctx, "aaa3", "pblob3", 2, 4, &tmp)) require.Equal(t, []byte{3, 4, 5, 6}, tmp.ToByteSlice()) } @@ -74,7 +74,7 @@ func TestContentCacheForData_Passthrough(t *testing.T) { ctx := testlogging.Context(t) - dataCache, err := newContentCacheForData(ctx, underlying, nil, cache.SweepSettings{ + dataCache, err := cache.NewContentCacheForData(ctx, underlying, nil, cache.SweepSettings{ MaxSizeBytes: 100, }, []byte{1, 2, 3, 4}) @@ -84,6 +84,6 @@ func TestContentCacheForData_Passthrough(t *testing.T) { var tmp gather.WriteBuffer defer tmp.Close() - require.NoError(t, dataCache.getContent(ctx, "key1", "blob1", 0, 5, &tmp)) + require.NoError(t, dataCache.GetContent(ctx, "key1", "blob1", 0, 5, &tmp)) require.Equal(t, []byte{1, 2, 3, 4, 5}, tmp.ToByteSlice()) } diff --git a/repo/content/content_cache_metadata.go b/internal/cache/content_cache_metadata.go similarity index 61% rename from repo/content/content_cache_metadata.go rename to internal/cache/content_cache_metadata.go index af75b93c3..0d360f835 100644 --- a/repo/content/content_cache_metadata.go +++ b/internal/cache/content_cache_metadata.go @@ -1,4 +1,4 @@ -package content +package cache import ( "context" @@ -8,55 +8,23 @@ import ( "github.com/pkg/errors" "go.opencensus.io/stats" - "golang.org/x/sync/errgroup" - "github.com/kopia/kopia/internal/cache" "github.com/kopia/kopia/internal/gather" "github.com/kopia/kopia/internal/impossible" "github.com/kopia/kopia/repo/blob" ) const ( - metadataCacheSyncParallelism = 16 - metadataCacheMutexShards = 256 + metadataCacheMutexShards = 256 ) type contentCacheForMetadata struct { - pc *cache.PersistentCache + pc *PersistentCache st blob.Storage shardedMutexes [metadataCacheMutexShards]sync.Mutex } -// sync synchronizes metadata cache with all blobs found in the storage. -func (c *contentCacheForMetadata) sync(ctx context.Context) error { - sem := make(chan struct{}, metadataCacheSyncParallelism) - - var eg errgroup.Group - - // list all blobs and fetch contents into cache in parallel. - if err := c.st.ListBlobs(ctx, PackBlobIDPrefixSpecial, func(bm blob.Metadata) error { - // acquire semaphore - sem <- struct{}{} - eg.Go(func() error { - defer func() { - <-sem - }() - - var tmp gather.WriteBuffer - defer tmp.Close() - - return c.getContent(ctx, "dummy", bm.BlobID, 0, 1, &tmp) - }) - - return nil - }); err != nil { - return errors.Wrap(err, "error listing blobs") - } - - return errors.Wrap(eg.Wait(), "error synchronizing metadata cache") -} - func (c *contentCacheForMetadata) mutexForBlob(blobID blob.ID) *sync.Mutex { // hash the blob ID to pick one of the sharded mutexes. h := fnv.New32() @@ -66,7 +34,7 @@ func (c *contentCacheForMetadata) mutexForBlob(blobID blob.ID) *sync.Mutex { return &c.shardedMutexes[mutexID] } -func (c *contentCacheForMetadata) getContent(ctx context.Context, contentID ID, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error { +func (c *contentCacheForMetadata) GetContent(ctx context.Context, contentID string, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error { // try getting from cache first if c.pc.GetPartial(ctx, string(blobID), offset, length, output) { return nil @@ -106,7 +74,7 @@ func (c *contentCacheForMetadata) getContent(ctx context.Context, contentID ID, return nil } -func (c *contentCacheForMetadata) prefetchBlob(ctx context.Context, blobID blob.ID) error { +func (c *contentCacheForMetadata) PrefetchBlob(ctx context.Context, blobID blob.ID) error { var blobData gather.WriteBuffer defer blobData.Close() @@ -126,13 +94,13 @@ func (c *contentCacheForMetadata) prefetchBlob(ctx context.Context, blobID blob. func (c *contentCacheForMetadata) fetchBlobInternal(ctx context.Context, blobID blob.ID, blobData *gather.WriteBuffer) error { // read the entire blob if err := c.st.GetBlob(ctx, blobID, 0, -1, blobData); err != nil { - stats.Record(ctx, cache.MetricMissErrors.M(1)) + stats.Record(ctx, MetricMissErrors.M(1)) // nolint:wrapcheck return err } - stats.Record(ctx, cache.MetricMissBytes.M(int64(blobData.Length()))) + stats.Record(ctx, MetricMissBytes.M(int64(blobData.Length()))) // store the whole blob in the cache. c.pc.Put(ctx, string(blobID), blobData.Bytes()) @@ -140,16 +108,21 @@ func (c *contentCacheForMetadata) fetchBlobInternal(ctx context.Context, blobID return nil } -func (c *contentCacheForMetadata) close(ctx context.Context) { +func (c *contentCacheForMetadata) Close(ctx context.Context) { c.pc.Close(ctx) } -func newContentCacheForMetadata(ctx context.Context, st blob.Storage, cacheStorage cache.Storage, sweep cache.SweepSettings) (contentCache, error) { +func (c *contentCacheForMetadata) CacheStorage() Storage { + return c.pc.cacheStorage +} + +// NewContentCacheForMetadata creates new content cache for metadata contents. +func NewContentCacheForMetadata(ctx context.Context, st blob.Storage, cacheStorage Storage, sweep SweepSettings) (ContentCache, error) { if cacheStorage == nil { return passthroughContentCache{st}, nil } - pc, err := cache.NewPersistentCache(ctx, "metadata cache", cacheStorage, cache.NoProtection(), sweep) + pc, err := NewPersistentCache(ctx, "metadata cache", cacheStorage, NoProtection(), sweep) if err != nil { return nil, errors.Wrap(err, "unable to create base cache") } diff --git a/internal/cache/content_cache_metadata_test.go b/internal/cache/content_cache_metadata_test.go new file mode 100644 index 000000000..75b72dd30 --- /dev/null +++ b/internal/cache/content_cache_metadata_test.go @@ -0,0 +1,81 @@ +package cache_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/kopia/kopia/internal/blobtesting" + "github.com/kopia/kopia/internal/cache" + "github.com/kopia/kopia/internal/gather" + "github.com/kopia/kopia/internal/testlogging" + "github.com/kopia/kopia/repo/blob" +) + +func TestContentCacheForMetadata(t *testing.T) { + ctx := testlogging.Context(t) + + underlyingData := blobtesting.DataMap{} + underlying := blobtesting.NewMapStorage(underlyingData, nil, nil) + + cacheData := blobtesting.DataMap{} + metadataCacheStorage := blobtesting.NewMapStorage(cacheData, nil, nil).(cache.Storage) + + metadataCache, err := cache.NewContentCacheForMetadata(ctx, underlying, metadataCacheStorage, cache.SweepSettings{ + MaxSizeBytes: 100, + }) + require.NoError(t, err) + + var tmp gather.WriteBuffer + defer tmp.Close() + + // get something we don't have in the underlying storage + require.ErrorIs(t, metadataCache.GetContent(ctx, "key1", "blob1", 0, 3, &tmp), blob.ErrBlobNotFound) + + require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) + + require.NoError(t, metadataCache.GetContent(ctx, "key1", "blob1", 0, -1, &tmp)) + require.Equal(t, []byte{1, 2, 3, 4, 5, 6}, tmp.ToByteSlice()) + + require.NoError(t, metadataCache.GetContent(ctx, "key1", "blob1", 0, 3, &tmp)) + require.Equal(t, []byte{1, 2, 3}, tmp.ToByteSlice()) + + // cache has the entire blob + require.Len(t, cacheData, 1) + + require.NoError(t, metadataCache.GetContent(ctx, "key1", "blob1", 3, 3, &tmp)) + require.Equal(t, []byte{4, 5, 6}, tmp.ToByteSlice()) + + // out of bounds + require.Error(t, metadataCache.GetContent(ctx, "key1", "blob1", 3, 9, &tmp)) + + // cache has the entire blob + require.Len(t, cacheData, 1) + + metadataCache.Close(ctx) + + // get slice with cache miss + require.NoError(t, underlying.PutBlob(ctx, "blob2", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) + require.NoError(t, metadataCache.GetContent(ctx, "key1", "blob2", 3, 3, &tmp)) + require.Equal(t, []byte{4, 5, 6}, tmp.ToByteSlice()) +} + +func TestContentCacheForMetadata_Passthrough(t *testing.T) { + underlyingData := blobtesting.DataMap{} + underlying := blobtesting.NewMapStorage(underlyingData, nil, nil) + + ctx := testlogging.Context(t) + + metadataCache, err := cache.NewContentCacheForMetadata(ctx, underlying, nil, cache.SweepSettings{ + MaxSizeBytes: 100, + }) + + require.NoError(t, err) + require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) + + var tmp gather.WriteBuffer + defer tmp.Close() + + require.NoError(t, metadataCache.GetContent(ctx, "key1", "blob1", 0, -1, &tmp)) + require.Equal(t, []byte{1, 2, 3, 4, 5, 6}, tmp.ToByteSlice()) +} diff --git a/internal/cache/content_cache_passthrough.go b/internal/cache/content_cache_passthrough.go new file mode 100644 index 000000000..05753414b --- /dev/null +++ b/internal/cache/content_cache_passthrough.go @@ -0,0 +1,32 @@ +package cache + +import ( + "context" + + "github.com/kopia/kopia/internal/gather" + "github.com/kopia/kopia/repo/blob" +) + +// passthroughContentCache is a contentCache which does no caching. +type passthroughContentCache struct { + st blob.Storage +} + +func (c passthroughContentCache) Close(ctx context.Context) {} + +func (c passthroughContentCache) GetContent(ctx context.Context, contentID string, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error { + // nolint:wrapcheck + return c.st.GetBlob(ctx, blobID, offset, length, output) +} + +func (c passthroughContentCache) PrefetchBlob(ctx context.Context, blobID blob.ID) error { + return nil +} + +func (c passthroughContentCache) Sync(ctx context.Context, blobPrefix blob.ID) error { + return nil +} + +func (c passthroughContentCache) CacheStorage() Storage { + return nil +} diff --git a/repo/content/content_cache_test.go b/internal/cache/content_cache_test.go similarity index 82% rename from repo/content/content_cache_test.go rename to internal/cache/content_cache_test.go index bad3838eb..22236569d 100644 --- a/repo/content/content_cache_test.go +++ b/internal/cache/content_cache_test.go @@ -1,4 +1,4 @@ -package content +package cache_test import ( "bytes" @@ -55,34 +55,27 @@ func TestCacheExpiration(t *testing.T) { underlyingStorage := newUnderlyingStorageForContentCacheTesting(t) - pc, err := cache.NewPersistentCache(testlogging.Context(t), "test cache", cacheStorage.(cache.Storage), cache.NoProtection(), cache.SweepSettings{ + ctx := testlogging.Context(t) + cc, err := cache.NewContentCacheForData(ctx, underlyingStorage, cacheStorage.(cache.Storage), cache.SweepSettings{ MaxSizeBytes: 10000, SweepFrequency: 500 * time.Millisecond, TouchThreshold: -1, - }) - if err != nil { - t.Fatalf("unable to create base cache: %v", err) - } + }, nil) - cc := &contentCacheForData{ - st: underlyingStorage, - pc: pc, - } + require.NoError(t, err) - ctx := testlogging.Context(t) - - defer cc.close(ctx) + defer cc.Close(ctx) var tmp gather.WriteBuffer defer tmp.Close() - err = cc.getContent(ctx, "00000a", "content-4k", 0, -1, &tmp) // 4k + err = cc.GetContent(ctx, "00000a", "content-4k", 0, -1, &tmp) // 4k require.NoError(t, err) - err = cc.getContent(ctx, "00000b", "content-4k", 0, -1, &tmp) // 4k + err = cc.GetContent(ctx, "00000b", "content-4k", 0, -1, &tmp) // 4k require.NoError(t, err) - err = cc.getContent(ctx, "00000c", "content-4k", 0, -1, &tmp) // 4k + err = cc.GetContent(ctx, "00000c", "content-4k", 0, -1, &tmp) // 4k require.NoError(t, err) - err = cc.getContent(ctx, "00000d", "content-4k", 0, -1, &tmp) // 4k + err = cc.GetContent(ctx, "00000d", "content-4k", 0, -1, &tmp) // 4k require.NoError(t, err) // wait for a sweep @@ -94,7 +87,7 @@ func TestCacheExpiration(t *testing.T) { require.NoError(t, underlyingStorage.DeleteBlob(ctx, "content-4k")) cases := []struct { - contentID ID + contentID string expectedError error }{ {"00000a", blob.ErrBlobNotFound}, @@ -104,7 +97,7 @@ func TestCacheExpiration(t *testing.T) { } for _, tc := range cases { - got := cc.getContent(ctx, tc.contentID, "content-4k", 0, -1, &tmp) + got := cc.GetContent(ctx, tc.contentID, "content-4k", 0, -1, &tmp) if want := tc.expectedError; !errors.Is(got, want) { t.Errorf("unexpected error when getting content %v: %v wanted %v", tc.contentID, got, want) } else { @@ -125,26 +118,26 @@ func TestDiskContentCache(t *testing.T) { t.Fatal(err) } - cc, err := newContentCacheForData(ctx, newUnderlyingStorageForContentCacheTesting(t), cacheStorage, cache.SweepSettings{ + cc, err := cache.NewContentCacheForData(ctx, newUnderlyingStorageForContentCacheTesting(t), cacheStorage, cache.SweepSettings{ MaxSizeBytes: maxBytes, }, nil) if err != nil { t.Fatalf("err: %v", err) } - defer cc.close(ctx) + defer cc.Close(ctx) verifyContentCache(t, cc, cacheStorage) } -func verifyContentCache(t *testing.T, cc contentCache, cacheStorage blob.Storage) { +func verifyContentCache(t *testing.T, cc cache.ContentCache, cacheStorage blob.Storage) { t.Helper() ctx := testlogging.Context(t) t.Run("GetContentContent", func(t *testing.T) { cases := []struct { - contentID ID + contentID string blobID blob.ID offset int64 length int64 @@ -167,7 +160,7 @@ func verifyContentCache(t *testing.T, cc contentCache, cacheStorage blob.Storage defer v.Close() for _, tc := range cases { - err := cc.getContent(ctx, tc.contentID, tc.blobID, tc.offset, tc.length, &v) + err := cc.GetContent(ctx, tc.contentID, tc.blobID, tc.offset, tc.length, &v) if (err != nil) != (tc.err != nil) { t.Errorf("unexpected error for %v: %+v, wanted %+v", tc.contentID, err, tc.err) } else if err != nil && err.Error() != tc.err.Error() { @@ -195,7 +188,7 @@ func verifyContentCache(t *testing.T, cc contentCache, cacheStorage blob.Storage require.NoError(t, cacheStorage.PutBlob(ctx, cacheKey, b, blob.PutOptions{})) - err := cc.getContent(ctx, "xf0f0f1", "content-1", 1, 5, &tmp) + err := cc.GetContent(ctx, "xf0f0f1", "content-1", 1, 5, &tmp) if err != nil { t.Fatalf("error in getContent: %v", err) } @@ -216,7 +209,7 @@ func TestCacheFailureToOpen(t *testing.T) { faultyCache.AddFault(blobtesting.MethodGetMetadata).ErrorInstead(someError) // Will fail because of ListBlobs failure. - _, err := newContentCacheForData(testlogging.Context(t), underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) + _, err := cache.NewContentCacheForData(testlogging.Context(t), underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) if err == nil || !strings.Contains(err.Error(), someError.Error()) { t.Errorf("invalid error %v, wanted: %v", err, someError) } @@ -224,12 +217,12 @@ func TestCacheFailureToOpen(t *testing.T) { // ListBlobs fails only once, next time it succeeds. ctx := testlogging.Context(t) - cc, err := newContentCacheForData(ctx, underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) + cc, err := cache.NewContentCacheForData(ctx, underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) if err != nil { t.Fatalf("err: %v", err) } - cc.close(ctx) + cc.Close(ctx) } func TestCacheFailureToWrite(t *testing.T) { @@ -240,21 +233,21 @@ func TestCacheFailureToWrite(t *testing.T) { underlyingStorage := newUnderlyingStorageForContentCacheTesting(t) faultyCache := blobtesting.NewFaultyStorage(cacheStorage) - cc, err := newContentCacheForData(testlogging.Context(t), underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) + cc, err := cache.NewContentCacheForData(testlogging.Context(t), underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) if err != nil { t.Fatalf("err: %v", err) } ctx := testlogging.Context(t) - defer cc.close(ctx) + defer cc.Close(ctx) faultyCache.AddFault(blobtesting.MethodPutBlob).ErrorInstead(someError) var v gather.WriteBuffer defer v.Close() - if err = cc.getContent(ctx, "aa", "content-1", 0, 3, &v); err != nil { + if err = cc.GetContent(ctx, "aa", "content-1", 0, 3, &v); err != nil { t.Errorf("write failure wasn't ignored: %v", err) } @@ -280,14 +273,14 @@ func TestCacheFailureToRead(t *testing.T) { underlyingStorage := newUnderlyingStorageForContentCacheTesting(t) faultyCache := blobtesting.NewFaultyStorage(cacheStorage) - cc, err := newContentCacheForData(testlogging.Context(t), underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) + cc, err := cache.NewContentCacheForData(testlogging.Context(t), underlyingStorage, withoutTouchBlob{faultyCache}, cache.SweepSettings{MaxSizeBytes: 10000}, nil) if err != nil { t.Fatalf("err: %v", err) } ctx := testlogging.Context(t) - defer cc.close(ctx) + defer cc.Close(ctx) faultyCache.AddFault(blobtesting.MethodGetBlob).ErrorInstead(someError).Repeat(100) @@ -295,7 +288,7 @@ func TestCacheFailureToRead(t *testing.T) { defer v.Close() for i := 0; i < 2; i++ { - require.NoError(t, cc.getContent(ctx, "aa", "content-1", 0, 3, &v)) + require.NoError(t, cc.GetContent(ctx, "aa", "content-1", 0, 3, &v)) if got, want := v.ToByteSlice(), []byte{1, 2, 3}; !reflect.DeepEqual(got, want) { t.Errorf("unexpected value retrieved from cache: %v, want: %v", got, want) diff --git a/repo/content/committed_read_manager.go b/repo/content/committed_read_manager.go index 00c734659..01e0c8eff 100644 --- a/repo/content/committed_read_manager.go +++ b/repo/content/committed_read_manager.go @@ -81,8 +81,8 @@ type SharedManager struct { indexBlobManagerV0 *indexBlobManagerV0 indexBlobManagerV1 *indexBlobManagerV1 - contentCache contentCache - metadataCache contentCache + contentCache cache.ContentCache + metadataCache cache.ContentCache committedContents *committedContentIndex crypter *Crypter enc *encryptedBlobMgr @@ -232,7 +232,7 @@ func (sm *SharedManager) loadPackIndexesLocked(ctx context.Context) error { return errors.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts) } -func (sm *SharedManager) getCacheForContentID(id ID) contentCache { +func (sm *SharedManager) getCacheForContentID(id ID) cache.ContentCache { if id.HasPrefix() { return sm.metadataCache } @@ -373,7 +373,7 @@ func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *Ca return errors.Wrap(err, "unable to initialize data cache storage") } - dataCache, err := newContentCacheForData(ctx, sm.st, dataCacheStorage, cache.SweepSettings{ + dataCache, err := cache.NewContentCacheForData(ctx, sm.st, dataCacheStorage, cache.SweepSettings{ MaxSizeBytes: caching.MaxCacheSizeBytes, MinSweepAge: caching.MinContentSweepAge.DurationOrDefault(DefaultDataCacheSweepAge), }, caching.HMACSecret) @@ -391,7 +391,7 @@ func (sm *SharedManager) setupReadManagerCaches(ctx context.Context, caching *Ca return errors.Wrap(err, "unable to initialize data cache storage") } - metadataCache, err := newContentCacheForMetadata(ctx, sm.st, metadataCacheStorage, cache.SweepSettings{ + metadataCache, err := cache.NewContentCacheForMetadata(ctx, sm.st, metadataCacheStorage, cache.SweepSettings{ MaxSizeBytes: metadataCacheSize, MinSweepAge: caching.MinMetadataSweepAge.DurationOrDefault(DefaultMetadataCacheSweepAge), }) @@ -508,8 +508,8 @@ func (sm *SharedManager) release(ctx context.Context) error { return errors.Wrap(err, "error closing committed content index") } - sm.contentCache.close(ctx) - sm.metadataCache.close(ctx) + sm.contentCache.Close(ctx) + sm.metadataCache.Close(ctx) if sm.internalLogger != nil { sm.internalLogger.Sync() // nolint:errcheck diff --git a/repo/content/content_cache.go b/repo/content/content_cache.go deleted file mode 100644 index 33e513658..000000000 --- a/repo/content/content_cache.go +++ /dev/null @@ -1,14 +0,0 @@ -package content - -import ( - "context" - - "github.com/kopia/kopia/internal/gather" - "github.com/kopia/kopia/repo/blob" -) - -type contentCache interface { - close(ctx context.Context) - getContent(ctx context.Context, contentID ID, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error - prefetchBlob(ctx context.Context, blobID blob.ID) error -} diff --git a/repo/content/content_cache_data.go b/repo/content/content_cache_data.go deleted file mode 100644 index a64799d52..000000000 --- a/repo/content/content_cache_data.go +++ /dev/null @@ -1,88 +0,0 @@ -package content - -import ( - "context" - - "github.com/pkg/errors" - "go.opencensus.io/stats" - - "github.com/kopia/kopia/internal/cache" - "github.com/kopia/kopia/internal/gather" - "github.com/kopia/kopia/repo/blob" -) - -type contentCacheForData struct { - pc *cache.PersistentCache - st blob.Storage -} - -func contentIDCacheKey(contentID ID) string { - // content IDs with odd length have a single-byte prefix. - // move the prefix to the end of cache key to make sure the top level shard is spread 256 ways. - if len(contentID)%2 == 1 { - return string(contentID[1:] + contentID[0:1]) - } - - return string(contentID) -} - -func blobIDCacheKey(id blob.ID) string { - return string(id[1:] + id[0:1]) -} - -func (c *contentCacheForData) getContent(ctx context.Context, contentID ID, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error { - if c.pc.GetPartial(ctx, blobIDCacheKey(blobID), offset, length, output) { - return nil - } - - output.Reset() - - // nolint:wrapcheck - return c.pc.GetOrLoad(ctx, contentIDCacheKey(contentID), func(output *gather.WriteBuffer) error { - // nolint:wrapcheck - return c.st.GetBlob(ctx, blobID, offset, length, output) - }, output) -} - -func (c *contentCacheForData) close(ctx context.Context) { - c.pc.Close(ctx) -} - -func (c contentCacheForData) prefetchBlob(ctx context.Context, blobID blob.ID) error { - var blobData gather.WriteBuffer - defer blobData.Close() - - if c.pc.GetPartial(ctx, blobIDCacheKey(blobID), 0, 1, &blobData) { - return nil - } - - // read the entire blob - if err := c.st.GetBlob(ctx, blobID, 0, -1, &blobData); err != nil { - stats.Record(ctx, cache.MetricMissErrors.M(1)) - // nolint:wrapcheck - return err - } - - stats.Record(ctx, cache.MetricMissBytes.M(int64(blobData.Length()))) - - // store the whole blob in the cache. - c.pc.Put(ctx, blobIDCacheKey(blobID), blobData.Bytes()) - - return nil -} - -func newContentCacheForData(ctx context.Context, st blob.Storage, cacheStorage cache.Storage, sweep cache.SweepSettings, hmacSecret []byte) (contentCache, error) { - if cacheStorage == nil { - return passthroughContentCache{st}, nil - } - - pc, err := cache.NewPersistentCache(ctx, "content cache", cacheStorage, cache.ChecksumProtection(hmacSecret), sweep) - if err != nil { - return nil, errors.Wrap(err, "unable to create base cache") - } - - return &contentCacheForData{ - st: st, - pc: pc, - }, nil -} diff --git a/repo/content/content_cache_metadata_test.go b/repo/content/content_cache_metadata_test.go deleted file mode 100644 index c9adf8920..000000000 --- a/repo/content/content_cache_metadata_test.go +++ /dev/null @@ -1,118 +0,0 @@ -package content - -import ( - "testing" - - "github.com/pkg/errors" - "github.com/stretchr/testify/require" - - "github.com/kopia/kopia/internal/blobtesting" - "github.com/kopia/kopia/internal/cache" - "github.com/kopia/kopia/internal/gather" - "github.com/kopia/kopia/internal/testlogging" - "github.com/kopia/kopia/repo/blob" -) - -func TestContentCacheForMetadata(t *testing.T) { - ctx := testlogging.Context(t) - - underlyingData := blobtesting.DataMap{} - underlying := blobtesting.NewMapStorage(underlyingData, nil, nil) - - cacheData := blobtesting.DataMap{} - metadataCacheStorage := blobtesting.NewMapStorage(cacheData, nil, nil).(cache.Storage) - - metadataCache, err := newContentCacheForMetadata(ctx, underlying, metadataCacheStorage, cache.SweepSettings{ - MaxSizeBytes: 100, - }) - require.NoError(t, err) - - var tmp gather.WriteBuffer - defer tmp.Close() - - // get something we don't have in the underlying storage - require.ErrorIs(t, metadataCache.getContent(ctx, "key1", "blob1", 0, 3, &tmp), blob.ErrBlobNotFound) - - require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - - require.NoError(t, metadataCache.getContent(ctx, "key1", "blob1", 0, -1, &tmp)) - require.Equal(t, []byte{1, 2, 3, 4, 5, 6}, tmp.ToByteSlice()) - - require.NoError(t, metadataCache.getContent(ctx, "key1", "blob1", 0, 3, &tmp)) - require.Equal(t, []byte{1, 2, 3}, tmp.ToByteSlice()) - - // cache has the entire blob - require.Len(t, cacheData, 1) - - require.NoError(t, metadataCache.getContent(ctx, "key1", "blob1", 3, 3, &tmp)) - require.Equal(t, []byte{4, 5, 6}, tmp.ToByteSlice()) - - // out of bounds - require.Error(t, metadataCache.getContent(ctx, "key1", "blob1", 3, 9, &tmp)) - - // cache has the entire blob - require.Len(t, cacheData, 1) - - metadataCache.close(ctx) - - // get slice with cache miss - require.NoError(t, underlying.PutBlob(ctx, "blob2", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, metadataCache.getContent(ctx, "key1", "blob2", 3, 3, &tmp)) - require.Equal(t, []byte{4, 5, 6}, tmp.ToByteSlice()) -} - -func TestContentCacheForMetadata_Passthrough(t *testing.T) { - underlyingData := blobtesting.DataMap{} - underlying := blobtesting.NewMapStorage(underlyingData, nil, nil) - - ctx := testlogging.Context(t) - - metadataCache, err := newContentCacheForMetadata(ctx, underlying, nil, cache.SweepSettings{ - MaxSizeBytes: 100, - }) - - require.NoError(t, err) - require.NoError(t, underlying.PutBlob(ctx, "blob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - - var tmp gather.WriteBuffer - defer tmp.Close() - - require.NoError(t, metadataCache.getContent(ctx, "key1", "blob1", 0, -1, &tmp)) - require.Equal(t, []byte{1, 2, 3, 4, 5, 6}, tmp.ToByteSlice()) -} - -func TestContentCacheForMetadata_Sync(t *testing.T) { - ctx := testlogging.Context(t) - - underlyingData := blobtesting.DataMap{} - underlying := blobtesting.NewMapStorage(underlyingData, nil, nil) - fs := blobtesting.NewFaultyStorage(underlying) - - cacheData := blobtesting.DataMap{} - metadataCacheStorage := blobtesting.NewMapStorage(cacheData, nil, nil).(cache.Storage) - - metadataCache, err := newContentCacheForMetadata(ctx, fs, metadataCacheStorage, cache.SweepSettings{ - MaxSizeBytes: 100, - }) - require.NoError(t, err) - - var tmp gather.WriteBuffer - defer tmp.Close() - - require.NoError(t, underlying.PutBlob(ctx, "qblob1", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, underlying.PutBlob(ctx, "qblob2", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, underlying.PutBlob(ctx, "qblob3", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, underlying.PutBlob(ctx, "qblob4", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, underlying.PutBlob(ctx, "zblob5", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - require.NoError(t, underlying.PutBlob(ctx, "zblob6", gather.FromSlice([]byte{1, 2, 3, 4, 5, 6}), blob.PutOptions{})) - - require.NoError(t, metadataCache.(*contentCacheForMetadata).sync(ctx)) - - // q blobs will be brought in, z won't - require.Len(t, cacheData, 4) - - someError := errors.Errorf("some error") - - fs.AddFault(blobtesting.MethodListBlobs).ErrorInstead(someError) - require.ErrorIs(t, metadataCache.(*contentCacheForMetadata).sync(ctx), someError) -} diff --git a/repo/content/content_cache_passthrough.go b/repo/content/content_cache_passthrough.go deleted file mode 100644 index dd74ebbd6..000000000 --- a/repo/content/content_cache_passthrough.go +++ /dev/null @@ -1,24 +0,0 @@ -package content - -import ( - "context" - - "github.com/kopia/kopia/internal/gather" - "github.com/kopia/kopia/repo/blob" -) - -// passthroughContentCache is a contentCache which does no caching. -type passthroughContentCache struct { - st blob.Storage -} - -func (c passthroughContentCache) close(ctx context.Context) {} - -func (c passthroughContentCache) getContent(ctx context.Context, contentID ID, blobID blob.ID, offset, length int64, output *gather.WriteBuffer) error { - // nolint:wrapcheck - return c.st.GetBlob(ctx, blobID, offset, length, output) -} - -func (c passthroughContentCache) prefetchBlob(ctx context.Context, blobID blob.ID) error { - return nil -} diff --git a/repo/content/content_manager.go b/repo/content/content_manager.go index bdcd702c5..7e4819c9a 100644 --- a/repo/content/content_manager.go +++ b/repo/content/content_manager.go @@ -14,6 +14,7 @@ import ( "github.com/pkg/errors" "go.opencensus.io/stats" + "github.com/kopia/kopia/internal/cache" "github.com/kopia/kopia/internal/clock" "github.com/kopia/kopia/internal/gather" "github.com/kopia/kopia/repo/blob" @@ -811,15 +812,9 @@ func (bm *WriteManager) unlock() { bm.mu.Unlock() } -// SyncMetadataCache synchronizes metadata cache with metadata blobs in storage. -func (bm *WriteManager) SyncMetadataCache(ctx context.Context) error { - if cm, ok := bm.metadataCache.(*contentCacheForMetadata); ok { - return cm.sync(ctx) - } - - bm.log.Debugf("metadata cache not enabled") - - return nil +// MetadataCache returns an instance of metadata cache. +func (bm *WriteManager) MetadataCache() cache.ContentCache { + return bm.metadataCache } // ManagerOptions are the optional parameters for manager creation. diff --git a/repo/content/content_manager_lock_free.go b/repo/content/content_manager_lock_free.go index 2b0e297bf..fd1737d6c 100644 --- a/repo/content/content_manager_lock_free.go +++ b/repo/content/content_manager_lock_free.go @@ -98,7 +98,7 @@ func (bm *WriteManager) getContentDataReadLocked(ctx context.Context, pp *pendin // should never happen return errors.Wrap(err, "error appending pending content data to buffer") } - } else if err := bm.getCacheForContentID(bi.GetContentID()).getContent(ctx, bi.GetContentID(), bi.GetPackBlobID(), int64(bi.GetPackOffset()), int64(bi.GetPackedLength()), &payload); err != nil { + } else if err := bm.getCacheForContentID(bi.GetContentID()).GetContent(ctx, string(bi.GetContentID()), bi.GetPackBlobID(), int64(bi.GetPackOffset()), int64(bi.GetPackedLength()), &payload); err != nil { return errors.Wrap(err, "error getting cached content") } diff --git a/repo/content/content_manager_test.go b/repo/content/content_manager_test.go index f959f6582..6455ccb63 100644 --- a/repo/content/content_manager_test.go +++ b/repo/content/content_manager_test.go @@ -2050,6 +2050,10 @@ func (s *contentManagerSuite) TestCompression_NonCompressibleData(t *testing.T) verifyContent(ctx, t, bm2, cid, nonCompressibleData) } +func contentIDCacheKey(id ID) string { + return cache.ContentIDCacheKey(string(id)) +} + func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { ctx := testlogging.Context(t) data := blobtesting.DataMap{} @@ -2084,8 +2088,8 @@ func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { require.Equal(t, blob2, getContentInfo(t, bm, id5).GetPackBlobID()) require.Equal(t, blob2, getContentInfo(t, bm, id6).GetPackBlobID()) - ccd := bm.contentCache.(*contentCacheForData) - ccm := bm.metadataCache.(*contentCacheForMetadata) + ccd := bm.contentCache + ccm := bm.metadataCache hints := []string{ "", "default", "contents", "blobs", "none", @@ -2102,10 +2106,10 @@ func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { input: []ID{id1, id2, id3, id4, id5, id6, "no-such-content"}, wantResult: []ID{id1, id2, id3, id4, id5, id6}, wantDataCacheKeys: map[string][]string{ - "": {blobIDCacheKey(blob1), blobIDCacheKey(blob2)}, - "default": {blobIDCacheKey(blob1), blobIDCacheKey(blob2)}, + "": {cache.BlobIDCacheKey(blob1), cache.BlobIDCacheKey(blob2)}, + "default": {cache.BlobIDCacheKey(blob1), cache.BlobIDCacheKey(blob2)}, "contents": {contentIDCacheKey(id1), contentIDCacheKey(id2), contentIDCacheKey(id3), contentIDCacheKey(id4), contentIDCacheKey(id5), contentIDCacheKey(id6)}, - "blobs": {blobIDCacheKey(blob1), blobIDCacheKey(blob2)}, + "blobs": {cache.BlobIDCacheKey(blob1), cache.BlobIDCacheKey(blob2)}, "none": {}, }, }, @@ -2117,7 +2121,7 @@ func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { "": {contentIDCacheKey(id1)}, "default": {contentIDCacheKey(id1)}, "contents": {contentIDCacheKey(id1)}, - "blobs": {blobIDCacheKey(blob1)}, + "blobs": {cache.BlobIDCacheKey(blob1)}, "none": {}, }, }, @@ -2129,7 +2133,7 @@ func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { "": {contentIDCacheKey(id1), contentIDCacheKey(id4)}, "default": {contentIDCacheKey(id1), contentIDCacheKey(id4)}, "contents": {contentIDCacheKey(id1), contentIDCacheKey(id4)}, - "blobs": {blobIDCacheKey(blob1), blobIDCacheKey(blob2)}, + "blobs": {cache.BlobIDCacheKey(blob1), cache.BlobIDCacheKey(blob2)}, "none": {}, }, }, @@ -2138,10 +2142,10 @@ func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { input: []ID{id1, id4, id5}, wantResult: []ID{id1, id4, id5}, wantDataCacheKeys: map[string][]string{ - "": {contentIDCacheKey(id1), blobIDCacheKey(blob2)}, - "default": {contentIDCacheKey(id1), blobIDCacheKey(blob2)}, + "": {contentIDCacheKey(id1), cache.BlobIDCacheKey(blob2)}, + "default": {contentIDCacheKey(id1), cache.BlobIDCacheKey(blob2)}, "contents": {contentIDCacheKey(id1), contentIDCacheKey(id4), contentIDCacheKey(id5)}, - "blobs": {blobIDCacheKey(blob1), blobIDCacheKey(blob2)}, + "blobs": {cache.BlobIDCacheKey(blob1), cache.BlobIDCacheKey(blob2)}, "none": {}, }, }, @@ -2155,16 +2159,16 @@ func (s *contentManagerSuite) TestPrefetchContent(t *testing.T) { tc := tc t.Run(tc.name, func(t *testing.T) { - wipeCache(t, ccd.pc.CacheStorage()) - wipeCache(t, ccm.pc.CacheStorage()) + wipeCache(t, ccd.CacheStorage()) + wipeCache(t, ccm.CacheStorage()) - require.Empty(t, allCacheKeys(t, ccd.pc.CacheStorage())) - require.Empty(t, allCacheKeys(t, ccm.pc.CacheStorage())) + require.Empty(t, allCacheKeys(t, ccd.CacheStorage())) + require.Empty(t, allCacheKeys(t, ccm.CacheStorage())) require.Equal(t, tc.wantResult, bm.PrefetchContents(ctx, tc.input, hint)) - require.ElementsMatch(t, tc.wantDataCacheKeys[hint], allCacheKeys(t, ccd.pc.CacheStorage())) + require.ElementsMatch(t, tc.wantDataCacheKeys[hint], allCacheKeys(t, ccd.CacheStorage())) for _, cid := range tc.wantResult { _, err := bm.GetContent(ctx, cid) diff --git a/repo/content/content_prefetch.go b/repo/content/content_prefetch.go index e81ff1847..4ebc8aa33 100644 --- a/repo/content/content_prefetch.go +++ b/repo/content/content_prefetch.go @@ -114,11 +114,11 @@ func (bm *WriteManager) PrefetchContents(ctx context.Context, contentIDs []ID, h for w := range workCh { switch { case strings.HasPrefix(string(w.blobID), string(PackBlobIDPrefixRegular)): - if err := bm.contentCache.prefetchBlob(ctx, w.blobID); err != nil { + if err := bm.contentCache.PrefetchBlob(ctx, w.blobID); err != nil { bm.log.Debugw("error prefetching data blob", "blobID", w.blobID, "err", err) } case strings.HasPrefix(string(w.blobID), string(PackBlobIDPrefixSpecial)): - if err := bm.metadataCache.prefetchBlob(ctx, w.blobID); err != nil { + if err := bm.metadataCache.PrefetchBlob(ctx, w.blobID); err != nil { bm.log.Debugw("error prefetching metadata blob", "blobID", w.blobID, "err", err) } case w.contentID != "":