restic/internal/repository/parallel.go

66 lines
1.7 KiB
Go
Raw Normal View History

2015-07-04 15:47:42 +00:00
package repository
import (
2017-06-04 09:16:55 +00:00
"context"
2015-07-04 15:47:42 +00:00
2017-07-23 12:21:03 +00:00
"github.com/restic/restic/internal/debug"
2017-07-24 15:42:25 +00:00
"github.com/restic/restic/internal/restic"
"golang.org/x/sync/errgroup"
2015-07-04 15:47:42 +00:00
)
// ParallelWorkFunc gets one file ID to work on. If an error is returned,
2017-06-04 09:16:55 +00:00
// processing stops. When the contect is cancelled the function should return.
type ParallelWorkFunc func(ctx context.Context, id string) error
2016-08-31 20:39:36 +00:00
// ParallelIDWorkFunc gets one restic.ID to work on. If an error is returned,
2017-06-04 09:16:55 +00:00
// processing stops. When the context is cancelled the function should return.
type ParallelIDWorkFunc func(ctx context.Context, id restic.ID) error
2015-07-04 15:47:42 +00:00
// FilesInParallel runs n workers of f in parallel, on the IDs that
// repo.List(t) yields. If f returns an error, the process is aborted and the
2015-07-04 15:47:42 +00:00
// first error is returned.
func FilesInParallel(ctx context.Context, repo restic.Lister, t restic.FileType, n int, f ParallelWorkFunc) error {
g, ctx := errgroup.WithContext(ctx)
2015-07-04 15:47:42 +00:00
ch := make(chan string, n)
g.Go(func() error {
defer close(ch)
return repo.List(ctx, t, func(fi restic.FileInfo) error {
select {
case <-ctx.Done():
case ch <- fi.Name:
}
return nil
})
})
2015-07-04 15:47:42 +00:00
for i := 0; i < n; i++ {
g.Go(func() error {
for name := range ch {
err := f(ctx, name)
if err != nil {
return err
2015-07-04 15:47:42 +00:00
}
}
return nil
})
2015-07-04 15:47:42 +00:00
}
return g.Wait()
2015-07-04 15:47:42 +00:00
}
2016-08-31 20:39:36 +00:00
// ParallelWorkFuncParseID converts a function that takes a restic.ID to a
// function that takes a string. Filenames that do not parse as a restic.ID
// are ignored.
func ParallelWorkFuncParseID(f ParallelIDWorkFunc) ParallelWorkFunc {
2017-06-04 09:16:55 +00:00
return func(ctx context.Context, s string) error {
2016-08-31 20:39:36 +00:00
id, err := restic.ParseID(s)
if err != nil {
2016-09-27 20:35:08 +00:00
debug.Log("invalid ID %q: %v", id, err)
return nil
}
2017-06-04 09:16:55 +00:00
return f(ctx, id)
}
}