From 26f85779be1a08d5f6371855abe237c8da3e53e7 Mon Sep 17 00:00:00 2001 From: Alexander Weiss Date: Sat, 28 Nov 2020 09:32:06 +0100 Subject: [PATCH] Parallelize ForAllSnapshots --- internal/checker/checker.go | 98 +++++++------------------------------ internal/restic/snapshot.go | 59 +++++++++++++++++++--- 2 files changed, 71 insertions(+), 86 deletions(-) diff --git a/internal/checker/checker.go b/internal/checker/checker.go index 9b7ddaad4..309f932f5 100644 --- a/internal/checker/checker.go +++ b/internal/checker/checker.go @@ -298,86 +298,6 @@ func (e Error) Error() string { return e.Err.Error() } -func loadTreeFromSnapshot(ctx context.Context, repo restic.Repository, id restic.ID) (restic.ID, error) { - sn, err := restic.LoadSnapshot(ctx, repo, id) - if err != nil { - debug.Log("error loading snapshot %v: %v", id, err) - return restic.ID{}, err - } - - if sn.Tree == nil { - debug.Log("snapshot %v has no tree", id) - return restic.ID{}, errors.Errorf("snapshot %v has no tree", id) - } - - return *sn.Tree, nil -} - -// loadSnapshotTreeIDs loads all snapshots from backend and returns the tree IDs. -func loadSnapshotTreeIDs(ctx context.Context, repo restic.Repository) (restic.IDs, []error) { - var trees struct { - IDs restic.IDs - sync.Mutex - } - - var errs struct { - errs []error - sync.Mutex - } - - // track spawned goroutines using wg, create a new context which is - // cancelled as soon as an error occurs. - wg, ctx := errgroup.WithContext(ctx) - - ch := make(chan restic.ID) - - // send list of index files through ch, which is closed afterwards - wg.Go(func() error { - defer close(ch) - return repo.List(ctx, restic.SnapshotFile, func(id restic.ID, size int64) error { - select { - case <-ctx.Done(): - return nil - case ch <- id: - } - return nil - }) - }) - - // a worker receives an index ID from ch, loads the snapshot and the tree, - // and adds the result to errs and trees. - worker := func() error { - for id := range ch { - debug.Log("load snapshot %v", id) - - treeID, err := loadTreeFromSnapshot(ctx, repo, id) - if err != nil { - errs.Lock() - errs.errs = append(errs.errs, err) - errs.Unlock() - continue - } - - debug.Log("snapshot %v has tree %v", id, treeID) - trees.Lock() - trees.IDs = append(trees.IDs, treeID) - trees.Unlock() - } - return nil - } - - for i := 0; i < defaultParallelism; i++ { - wg.Go(worker) - } - - err := wg.Wait() - if err != nil { - errs.errs = append(errs.errs, err) - } - - return trees.IDs, errs.errs -} - // TreeError collects several errors that occurred while processing a tree. type TreeError struct { ID restic.ID @@ -586,6 +506,24 @@ func (c *Checker) filterTrees(ctx context.Context, backlog restic.IDs, loaderCha } } +func loadSnapshotTreeIDs(ctx context.Context, repo restic.Repository) (ids restic.IDs, errs []error) { + err := restic.ForAllSnapshots(ctx, repo, nil, func(id restic.ID, sn *restic.Snapshot, err error) error { + if err != nil { + errs = append(errs, err) + return nil + } + treeID := *sn.Tree + debug.Log("snapshot %v has tree %v", id, treeID) + ids = append(ids, treeID) + return nil + }) + if err != nil { + errs = append(errs, err) + } + + return ids, errs +} + // Structure checks that for all snapshots all referenced data blobs and // subtrees are available in the index. errChan is closed after all trees have // been traversed. diff --git a/internal/restic/snapshot.go b/internal/restic/snapshot.go index 75d05e7c4..584737d41 100644 --- a/internal/restic/snapshot.go +++ b/internal/restic/snapshot.go @@ -5,8 +5,11 @@ import ( "fmt" "os/user" "path/filepath" + "sync" "time" + "golang.org/x/sync/errgroup" + "github.com/restic/restic/internal/debug" ) @@ -66,17 +69,61 @@ func LoadSnapshot(ctx context.Context, repo Repository, id ID) (*Snapshot, error return sn, nil } -// ForAllSnapshots reads all snapshots and calls the given function. +const loadSnapshotParallelism = 5 + +// ForAllSnapshots reads all snapshots in parallel and calls the +// given function. It is guaranteed that the function is not run concurrently. +// If the called function returns an error, this function is cancelled and +// also returns this error. // If a snapshot ID is in excludeIDs, it will be ignored. func ForAllSnapshots(ctx context.Context, repo Repository, excludeIDs IDSet, fn func(ID, *Snapshot, error) error) error { - return repo.List(ctx, SnapshotFile, func(id ID, size int64) error { - if excludeIDs.Has(id) { + var m sync.Mutex + + // track spawned goroutines using wg, create a new context which is + // cancelled as soon as an error occurs. + wg, ctx := errgroup.WithContext(ctx) + + ch := make(chan ID) + + // send list of snapshot files through ch, which is closed afterwards + wg.Go(func() error { + defer close(ch) + return repo.List(ctx, SnapshotFile, func(id ID, size int64) error { + if excludeIDs.Has(id) { + return nil + } + + select { + case <-ctx.Done(): + return nil + case ch <- id: + } return nil - } - sn, err := LoadSnapshot(ctx, repo, id) - return fn(id, sn, err) + }) }) + // a worker receives an snapshot ID from ch, loads the snapshot + // and runs fn with id, the snapshot and the error + worker := func() error { + for id := range ch { + debug.Log("load snapshot %v", id) + sn, err := LoadSnapshot(ctx, repo, id) + + m.Lock() + err = fn(id, sn, err) + m.Unlock() + if err != nil { + return err + } + } + return nil + } + + for i := 0; i < loadSnapshotParallelism; i++ { + wg.Go(worker) + } + + return wg.Wait() } // LoadAllSnapshots returns a list of all snapshots in the repo.