diff --git a/block/block_cache.go b/block/block_cache.go index 9079597c4..99f5d6455 100644 --- a/block/block_cache.go +++ b/block/block_cache.go @@ -105,21 +105,18 @@ func (c *blockCache) sweepDirectory(ctx context.Context) (err error) { t0 := time.Now() log.Debug().Msg("sweeping cache") - ctx, cancel := context.WithCancel(ctx) - ch := c.cacheStorage.ListBlocks(ctx, "") - defer cancel() - var items []storage.BlockMetadata - - for it := range ch { - if it.Error != nil { - return fmt.Errorf("error listing cache: %v", it.Error) - } + err = c.cacheStorage.ListBlocks(ctx, "", func(it storage.BlockMetadata) error { items = append(items, it) + return nil + }) + + if err != nil { + return fmt.Errorf("error listing cache: %v", err) } sort.Slice(items, func(i, j int) bool { - return items[i].TimeStamp.After(items[j].TimeStamp) + return items[i].Timestamp.After(items[j].Timestamp) }) var totalRetainedSize int64 diff --git a/block/block_manager.go b/block/block_manager.go index e7b8d1242..3a38cf7de 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -871,22 +871,21 @@ func (bm *Manager) FindUnreferencedStorageFiles(ctx context.Context) ([]storage. } usedPackBlocks := findPackBlocksInUse(infos) - ch := bm.st.ListBlocks(ctx, PackBlockPrefix) - var unused []storage.BlockMetadata - for bi := range ch { - if bi.Error != nil { - return nil, fmt.Errorf("error listing storage blocks: %v", bi.Error) - } - + err = bm.st.ListBlocks(ctx, PackBlockPrefix, func(bi storage.BlockMetadata) error { u := usedPackBlocks[bi.BlockID] if u > 0 { log.Printf("pack %v, in use by %v blocks", bi.BlockID, u) - continue + return nil } unused = append(unused, bi) + return nil + }) + if err != nil { + return nil, fmt.Errorf("error listing storage blocks: %v", err) } + return unused, nil } @@ -1059,31 +1058,18 @@ type cachedList struct { // if 'full' is false, the function returns only blocks from the last 2 compactions. // The list of blocks is not guaranteed to be sorted. func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage) ([]IndexInfo, error) { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - ch := st.ListBlocks(ctx, newIndexBlockPrefix) - var results []IndexInfo - for it := range ch { - if it.Error != nil { - return nil, it.Error - } - + err := st.ListBlocks(ctx, newIndexBlockPrefix, func(it storage.BlockMetadata) error { ii := IndexInfo{ FileName: it.BlockID, - Timestamp: it.TimeStamp, + Timestamp: it.Timestamp, Length: it.Length, } results = append(results, ii) - } + return nil + }) - select { - case <-ctx.Done(): - return nil, ctx.Err() - default: - return results, nil - } + return results, err } // NewManager creates new block manager with given packing options and a formatter. diff --git a/cli/command_repository_create.go b/cli/command_repository_create.go index 3653ba7be..abffa6c7a 100644 --- a/cli/command_repository_create.go +++ b/cli/command_repository_create.go @@ -2,6 +2,7 @@ package cli import ( "context" + "errors" "fmt" "os" @@ -44,22 +45,17 @@ func newRepositoryOptionsFromFlags() *repo.NewRepositoryOptions { } func ensureEmpty(ctx context.Context, s storage.Storage) error { - ctx, cancel := context.WithCancel(ctx) - ch := s.ListBlocks(ctx, "") - d, hasData := <-ch - cancel() - - if hasData { - if d.Error != nil { - return d.Error - } - + hasDataError := errors.New("has data") + err := s.ListBlocks(ctx, "", func(cb storage.BlockMetadata) error { + return hasDataError + }) + if err == hasDataError { if !*createOverwrite { return fmt.Errorf("found existing data in storage, specify --overwrite to use anyway") } } - return nil + return err } func runCreateCommandWithStorage(ctx context.Context, st storage.Storage) error { diff --git a/cli/command_storage_list.go b/cli/command_storage_list.go index 59c6088cc..b638160df 100644 --- a/cli/command_storage_list.go +++ b/cli/command_storage_list.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/kopia/kopia/repo" + "github.com/kopia/kopia/storage" ) var ( @@ -15,27 +16,18 @@ var ( ) func runListStorageBlocks(ctx context.Context, rep *repo.Repository) error { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - ch := rep.Storage.ListBlocks(ctx, *storageListPrefix) - - for b := range ch { - if b.Error != nil { - return b.Error - } - + return rep.Storage.ListBlocks(ctx, *storageListPrefix, func(b storage.BlockMetadata) error { if *storageListMaxSize != 0 && b.Length > *storageListMaxSize { - continue + return nil } if *storageListMinSize != 0 && b.Length < *storageListMinSize { - continue + return nil } - fmt.Printf("%-70v %10v %v\n", b.BlockID, b.Length, b.TimeStamp.Local().Format(timeFormat)) - } - - return nil + fmt.Printf("%-70v %10v %v\n", b.BlockID, b.Length, b.Timestamp.Local().Format(timeFormat)) + return nil + }) } func init() { diff --git a/internal/storagetesting/asserts.go b/internal/storagetesting/asserts.go index 12c91e120..db1f4a8db 100644 --- a/internal/storagetesting/asserts.go +++ b/internal/storagetesting/asserts.go @@ -37,11 +37,11 @@ func AssertGetBlockNotFound(ctx context.Context, t *testing.T, s storage.Storage func AssertListResults(ctx context.Context, t *testing.T, s storage.Storage, prefix string, expected ...string) { var names []string - ctx, cancel := context.WithCancel(ctx) - blocks := s.ListBlocks(ctx, prefix) - defer cancel() - for e := range blocks { + if err := s.ListBlocks(ctx, prefix, func(e storage.BlockMetadata) error { names = append(names, e.BlockID) + return nil + }); err != nil { + t.Fatalf("err: %v", err) } if !reflect.DeepEqual(names, expected) { diff --git a/internal/storagetesting/map.go b/internal/storagetesting/map.go index 18f46bd28..40e94e209 100644 --- a/internal/storagetesting/map.go +++ b/internal/storagetesting/map.go @@ -59,36 +59,30 @@ func (s *mapStorage) DeleteBlock(ctx context.Context, id string) error { return nil } -func (s *mapStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata { - ch := make(chan storage.BlockMetadata) - go func() { - defer close(ch) - s.mutex.RLock() - defer s.mutex.RUnlock() +func (s *mapStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { + s.mutex.RLock() + defer s.mutex.RUnlock() - keys := []string{} - for k := range s.data { - if strings.HasPrefix(k, prefix) { - keys = append(keys, k) - } + keys := []string{} + for k := range s.data { + if strings.HasPrefix(k, prefix) { + keys = append(keys, k) } + } - sort.Strings(keys) + sort.Strings(keys) - for _, k := range keys { - v := s.data[k] - select { - case <-ctx.Done(): - return - case ch <- storage.BlockMetadata{ - BlockID: k, - Length: int64(len(v)), - TimeStamp: s.keyTime[k], - }: - } + for _, k := range keys { + v := s.data[k] + if err := callback(storage.BlockMetadata{ + BlockID: k, + Length: int64(len(v)), + Timestamp: s.keyTime[k], + }); err != nil { + return err } - }() - return ch + } + return nil } func (s *mapStorage) Close(ctx context.Context) error { diff --git a/internal/storagetesting/verify.go b/internal/storagetesting/verify.go index af9612bc6..6e67a1988 100644 --- a/internal/storagetesting/verify.go +++ b/internal/storagetesting/verify.go @@ -34,27 +34,5 @@ func VerifyStorage(ctx context.Context, t *testing.T, r storage.Storage) { AssertGetBlock(ctx, t, r, b.blk, b.contents) } - // List - ctx, cancel := context.WithCancel(ctx) - ch := r.ListBlocks(ctx, "ab") - defer cancel() - e1 := verifyNextBlock(t, blocks[0].blk, ch) - e2 := verifyNextBlock(t, blocks[2].blk, ch) - e3 := verifyNextBlock(t, blocks[3].blk, ch) - e4, ok := <-ch - if ok { - t.Errorf("unexpected item: %v", e4) - } - - if e1.TimeStamp.After(e2.TimeStamp) || e2.TimeStamp.After(e3.TimeStamp) { - t.Errorf("timings are not sorted: %v %v %v", e1.TimeStamp, e2.TimeStamp, e3.TimeStamp) - } -} - -func verifyNextBlock(t *testing.T, expectedBlockID string, ch <-chan storage.BlockMetadata) storage.BlockMetadata { - bm, ok := <-ch - if !ok || bm.BlockID != expectedBlockID { - t.Errorf("missing result: %v", expectedBlockID) - } - return bm + AssertListResults(ctx, t, r, "ab", blocks[0].blk, blocks[2].blk, blocks[3].blk) } diff --git a/storage/filesystem/filesystem_storage.go b/storage/filesystem/filesystem_storage.go index 53012d919..50dd2c2db 100644 --- a/storage/filesystem/filesystem_storage.go +++ b/storage/filesystem/filesystem_storage.go @@ -65,53 +65,48 @@ func makeFileName(blockID string) string { return blockID + fsStorageChunkSuffix } -func (fs *fsStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata { - result := make(chan storage.BlockMetadata) +func (fs *fsStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { + var walkDir func(string, string) error - var walkDir func(string, string) + walkDir = func(directory string, currentPrefix string) error { + entries, err := ioutil.ReadDir(directory) + if err != nil { + return err + } - walkDir = func(directory string, currentPrefix string) { - if entries, err := ioutil.ReadDir(directory); err == nil { - //log.Println("Walking", directory, "looking for", prefix) + for _, e := range entries { + if e.IsDir() { + newPrefix := currentPrefix + e.Name() + var match bool - for _, e := range entries { - if e.IsDir() { - newPrefix := currentPrefix + e.Name() - var match bool + if len(prefix) > len(newPrefix) { + match = strings.HasPrefix(prefix, newPrefix) + } else { + match = strings.HasPrefix(newPrefix, prefix) + } - if len(prefix) > len(newPrefix) { - match = strings.HasPrefix(prefix, newPrefix) - } else { - match = strings.HasPrefix(newPrefix, prefix) + if match { + if err := walkDir(directory+"/"+e.Name(), currentPrefix+e.Name()); err != nil { + return err } - - if match { - walkDir(directory+"/"+e.Name(), currentPrefix+e.Name()) - } - } else if fullID, ok := getstringFromFileName(currentPrefix + e.Name()); ok { - if strings.HasPrefix(fullID, prefix) { - select { - case <-ctx.Done(): - return - case result <- storage.BlockMetadata{ - BlockID: fullID, - Length: e.Size(), - TimeStamp: e.ModTime(), - }: - } + } + } else if fullID, ok := getstringFromFileName(currentPrefix + e.Name()); ok { + if strings.HasPrefix(fullID, prefix) { + if err := callback(storage.BlockMetadata{ + BlockID: fullID, + Length: e.Size(), + Timestamp: e.ModTime(), + }); err != nil { + return err } } } } + + return nil } - walkDirAndClose := func(directory string) { - walkDir(directory, "") - close(result) - } - - go walkDirAndClose(fs.Path) - return result + return walkDir(fs.Path, "") } func (fs *fsStorage) PutBlock(ctx context.Context, blockID string, data []byte) error { diff --git a/storage/gcs/gcs_storage.go b/storage/gcs/gcs_storage.go index 906fa704d..411c83e5c 100644 --- a/storage/gcs/gcs_storage.go +++ b/storage/gcs/gcs_storage.go @@ -126,42 +126,28 @@ func (gcs *gcsStorage) getObjectNameString(blockID string) string { return gcs.Prefix + blockID } -func (gcs *gcsStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata { - ch := make(chan storage.BlockMetadata, 100) +func (gcs *gcsStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { + lst := gcs.bucket.Objects(gcs.ctx, &gcsclient.Query{ + Prefix: gcs.getObjectNameString(prefix), + }) - go func() { - defer close(ch) - - lst := gcs.bucket.Objects(gcs.ctx, &gcsclient.Query{ - Prefix: gcs.getObjectNameString(prefix), - }) - - oa, err := lst.Next() - for err == nil { - bm := storage.BlockMetadata{ - BlockID: oa.Name[len(gcs.Prefix):], - Length: oa.Size, - TimeStamp: oa.Created, - } - select { - case ch <- bm: - case <-ctx.Done(): - return - } - oa, err = lst.Next() + oa, err := lst.Next() + for err == nil { + if err = callback(storage.BlockMetadata{ + BlockID: oa.Name[len(gcs.Prefix):], + Length: oa.Size, + Timestamp: oa.Created, + }); err != nil { + return err } + oa, err = lst.Next() + } - if err != iterator.Done { - select { - case ch <- storage.BlockMetadata{Error: translateError(err)}: - return - case <-ctx.Done(): - return - } - } - }() + if err != iterator.Done { + return err + } - return ch + return nil } func (gcs *gcsStorage) ConnectionInfo() storage.ConnectionInfo { diff --git a/storage/logging/logging_storage.go b/storage/logging/logging_storage.go index 040c0493c..d0b79bd09 100644 --- a/storage/logging/logging_storage.go +++ b/storage/logging/logging_storage.go @@ -43,11 +43,15 @@ func (s *loggingStorage) DeleteBlock(ctx context.Context, id string) error { return err } -func (s *loggingStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata { +func (s *loggingStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { t0 := time.Now() - ch := s.base.ListBlocks(ctx, prefix) - s.printf(s.prefix+"ListBlocks(%q) took %v", prefix, time.Since(t0)) - return ch + cnt := 0 + err := s.base.ListBlocks(ctx, prefix, func(bi storage.BlockMetadata) error { + cnt++ + return callback(bi) + }) + s.printf(s.prefix+"ListBlocks(%q)=%v returned %v items and took %v", prefix, err, cnt, time.Since(t0)) + return err } func (s *loggingStorage) Close(ctx context.Context) error { diff --git a/storage/s3/s3_storage.go b/storage/s3/s3_storage.go index 75803e45e..49c54d18e 100644 --- a/storage/s3/s3_storage.go +++ b/storage/s3/s3_storage.go @@ -120,38 +120,25 @@ func (s *s3Storage) getObjectNameString(b string) string { return s.Prefix + b } -func (s *s3Storage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata { - ch := make(chan storage.BlockMetadata, 100) - - go func() { - defer close(ch) - - oi := s.cli.ListObjects(s.BucketName, s.Prefix+prefix, false, ctx.Done()) - for o := range oi { - if err := o.Err; err != nil { - select { - case ch <- storage.BlockMetadata{Error: translateError(err)}: - return - case <-ctx.Done(): - return - } - } - - bm := storage.BlockMetadata{ - BlockID: o.Key[len(s.Prefix):], - Length: o.Size, - TimeStamp: o.LastModified, - } - - select { - case ch <- bm: - case <-ctx.Done(): - return - } +func (s *s3Storage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { + oi := s.cli.ListObjects(s.BucketName, s.Prefix+prefix, false, ctx.Done()) + for o := range oi { + if err := o.Err; err != nil { + return err } - }() - return ch + bm := storage.BlockMetadata{ + BlockID: o.Key[len(s.Prefix):], + Length: o.Size, + Timestamp: o.LastModified, + } + + if err := callback(bm); err != nil { + return err + } + } + + return nil } func (s *s3Storage) ConnectionInfo() storage.ConnectionInfo { diff --git a/storage/s3/s3_storage_test.go b/storage/s3/s3_storage_test.go index 19736319d..baa216eff 100644 --- a/storage/s3/s3_storage_test.go +++ b/storage/s3/s3_storage_test.go @@ -11,6 +11,7 @@ import ( "time" "github.com/kopia/kopia/internal/storagetesting" + "github.com/kopia/kopia/storage" "github.com/minio/minio-go" ) @@ -85,16 +86,8 @@ func cleanupOldData(ctx context.Context, t *testing.T) { t.Fatalf("err: %v", err) } - ctx, cancel := context.WithCancel(ctx) - items := st.ListBlocks(ctx, "") - defer cancel() - for it := range items { - if it.Error != nil { - t.Errorf("can't cleanup: %v", it.Error) - return - } - - age := time.Since(it.TimeStamp) + st.ListBlocks(ctx, "", func(it storage.BlockMetadata) error { + age := time.Since(it.Timestamp) if age > cleanupAge { if err := st.DeleteBlock(ctx, it.BlockID); err != nil { t.Errorf("warning: unable to delete %q: %v", it.BlockID, err) @@ -102,5 +95,6 @@ func cleanupOldData(ctx context.Context, t *testing.T) { } else { log.Printf("keeping %v", it.BlockID) } - } + return nil + }) } diff --git a/storage/storage.go b/storage/storage.go index 38a241147..0725466cd 100644 --- a/storage/storage.go +++ b/storage/storage.go @@ -35,7 +35,7 @@ type Storage interface { // ListBlocks returns a channel of BlockMetadata that describes storage blocks with existing name prefixes. // Iteration continues until all blocks have been listed or until client code invokes the returned cancellation function. - ListBlocks(ctx context.Context, prefix string) <-chan BlockMetadata + ListBlocks(ctx context.Context, prefix string, cb func(bm BlockMetadata) error) error // ConnectionInfo returns JSON-serializable data structure containing information required to // connect to storage. @@ -46,12 +46,10 @@ type Storage interface { } // BlockMetadata represents metadata about a single block in a storage. -// If Error field is set, no other field values should be assumed to be correct. type BlockMetadata struct { BlockID string Length int64 - TimeStamp time.Time - Error error + Timestamp time.Time } // ErrBlockNotFound is returned when a block cannot be found in storage. diff --git a/storage/webdav/webdav_storage.go b/storage/webdav/webdav_storage.go index d01e3a055..659a29e01 100644 --- a/storage/webdav/webdav_storage.go +++ b/storage/webdav/webdav_storage.go @@ -74,51 +74,48 @@ func makeFileName(blockID string) string { return blockID + fsStorageChunkSuffix } -func (d *davStorage) ListBlocks(ctx context.Context, prefix string) <-chan storage.BlockMetadata { - result := make(chan storage.BlockMetadata) +func (d *davStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { + var walkDir func(string, string) error - var walkDir func(string, string) + walkDir = func(urlStr string, currentPrefix string) error { + entries, err := d.propFindChildren(urlStr) + if err != nil { + return err + } - walkDir = func(urlStr string, currentPrefix string) { - if entries, err := d.propFindChildren(urlStr); err == nil { - for _, e := range entries { - if e.isCollection { - newPrefix := currentPrefix + e.name - var match bool + for _, e := range entries { + if e.isCollection { + newPrefix := currentPrefix + e.name + var match bool - if len(prefix) > len(newPrefix) { - match = strings.HasPrefix(prefix, newPrefix) - } else { - match = strings.HasPrefix(newPrefix, prefix) + if len(prefix) > len(newPrefix) { + match = strings.HasPrefix(prefix, newPrefix) + } else { + match = strings.HasPrefix(newPrefix, prefix) + } + + if match { + if err := walkDir(urlStr+"/"+e.name, currentPrefix+e.name); err != nil { + return err } - - if match { - walkDir(urlStr+"/"+e.name, currentPrefix+e.name) - } - } else if fullID, ok := getBlockIDFromFileName(currentPrefix + e.name); ok { - if strings.HasPrefix(fullID, prefix) { - select { - case <-ctx.Done(): - return - case result <- storage.BlockMetadata{ - BlockID: fullID, - Length: e.length, - TimeStamp: e.modTime, - }: - } + } + } else if fullID, ok := getBlockIDFromFileName(currentPrefix + e.name); ok { + if strings.HasPrefix(fullID, prefix) { + if err := callback(storage.BlockMetadata{ + BlockID: fullID, + Length: e.length, + Timestamp: e.modTime, + }); err != nil { + return err } } } } + + return nil } - walkDirAndClose := func(urlStr string) { - walkDir(urlStr, "") - close(result) - } - - go walkDirAndClose(d.URL) - return result + return walkDir(d.URL, "") } func (d *davStorage) makeCollectionAll(urlStr string) error {