octoleo/restic
2
2
Fork 0
mirror of https://github.com/octoleo/restic.git synced 2024-11-25 22:27:35 +00:00
Code Issues Releases Wiki Activity

Add context to restic packages

Browse Source
This commit is contained in:
Alexander Neumann 2017-06-04 11:16:55 +02:00
parent 16fcd07110
commit cf497c2728
50 changed files with 432 additions and 422 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/cmds/restic/cmd_backup.go
View File

@ -2,6 +2,7 @@ package main
import (
"bufio"
"context"
"fmt"
"io"
"os"
@ -263,7 +264,7 @@ func readBackupFromStdin(opts BackupOptions, gopts GlobalOptions, args []string)
return err
}
err = repo.LoadIndex()
err = repo.LoadIndex(context.TODO())
if err != nil {
return err
}
@ -274,7 +275,7 @@ func readBackupFromStdin(opts BackupOptions, gopts GlobalOptions, args []string)
Hostname: opts.Hostname,
}
_, id, err := r.Archive(opts.StdinFilename, os.Stdin, newArchiveStdinProgress(gopts))
_, id, err := r.Archive(context.TODO(), opts.StdinFilename, os.Stdin, newArchiveStdinProgress(gopts))
if err != nil {
return err
}
@ -372,7 +373,7 @@ func runBackup(opts BackupOptions, gopts GlobalOptions, args []string) error {
return err
}
err = repo.LoadIndex()
err = repo.LoadIndex(context.TODO())
if err != nil {
return err
}
@ -391,7 +392,7 @@ func runBackup(opts BackupOptions, gopts GlobalOptions, args []string) error {
// Find last snapshot to set it as parent, if not already set
if !opts.Force && parentSnapshotID == nil {
id, err := restic.FindLatestSnapshot(repo, target, opts.Tags, opts.Hostname)
id, err := restic.FindLatestSnapshot(context.TODO(), repo, target, opts.Tags, opts.Hostname)
if err == nil {
parentSnapshotID = &id
} else if err != restic.ErrNoSnapshotFound {
@ -489,7 +490,7 @@ func runBackup(opts BackupOptions, gopts GlobalOptions, args []string) error {
Warnf("%s\rwarning for %s: %v\n", ClearLine(), dir, err)
}
_, id, err := arch.Snapshot(newArchiveProgress(gopts, stat), target, opts.Tags, opts.Hostname, parentSnapshotID)
_, id, err := arch.Snapshot(context.TODO(), newArchiveProgress(gopts, stat), target, opts.Tags, opts.Hostname, parentSnapshotID)
if err != nil {
return err
}

15
src/cmds/restic/cmd_cat.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"encoding/json"
"fmt"
"os"
@ -73,7 +74,7 @@ func runCat(gopts GlobalOptions, args []string) error {
fmt.Println(string(buf))
return nil
case "index":
buf, err := repo.LoadAndDecrypt(restic.IndexFile, id)
buf, err := repo.LoadAndDecrypt(context.TODO(), restic.IndexFile, id)
if err != nil {
return err
}
@ -83,7 +84,7 @@ func runCat(gopts GlobalOptions, args []string) error {
case "snapshot":
sn := &restic.Snapshot{}
err = repo.LoadJSONUnpacked(restic.SnapshotFile, id, sn)
err = repo.LoadJSONUnpacked(context.TODO(), restic.SnapshotFile, id, sn)
if err != nil {
return err
}
@ -98,7 +99,7 @@ func runCat(gopts GlobalOptions, args []string) error {
return nil
case "key":
h := restic.Handle{Type: restic.KeyFile, Name: id.String()}
buf, err := backend.LoadAll(repo.Backend(), h)
buf, err := backend.LoadAll(context.TODO(), repo.Backend(), h)
if err != nil {
return err
}
@ -125,7 +126,7 @@ func runCat(gopts GlobalOptions, args []string) error {
fmt.Println(string(buf))
return nil
case "lock":
lock, err := restic.LoadLock(repo, id)
lock, err := restic.LoadLock(context.TODO(), repo, id)
if err != nil {
return err
}
@ -141,7 +142,7 @@ func runCat(gopts GlobalOptions, args []string) error {
}
// load index, handle all the other types
err = repo.LoadIndex()
err = repo.LoadIndex(context.TODO())
if err != nil {
return err
}
@ -149,7 +150,7 @@ func runCat(gopts GlobalOptions, args []string) error {
switch tpe {
case "pack":
h := restic.Handle{Type: restic.DataFile, Name: id.String()}
buf, err := backend.LoadAll(repo.Backend(), h)
buf, err := backend.LoadAll(context.TODO(), repo.Backend(), h)
if err != nil {
return err
}
@ -171,7 +172,7 @@ func runCat(gopts GlobalOptions, args []string) error {
blob := list[0]
buf := make([]byte, blob.Length)
n, err := repo.LoadBlob(t, id, buf)
n, err := repo.LoadBlob(context.TODO(), t, id, buf)
if err != nil {
return err
}

12
src/cmds/restic/cmd_check.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"os"
"time"
@ -92,7 +93,7 @@ func runCheck(opts CheckOptions, gopts GlobalOptions, args []string) error {
chkr := checker.New(repo)
Verbosef("Load indexes\n")
hints, errs := chkr.LoadIndex()
hints, errs := chkr.LoadIndex(context.TODO())
dupFound := false
for _, hint := range hints {
@ -113,14 +114,11 @@ func runCheck(opts CheckOptions, gopts GlobalOptions, args []string) error {
return errors.Fatal("LoadIndex returned errors")
}
done := make(chan struct{})
defer close(done)
errorsFound := false
errChan := make(chan error)
Verbosef("Check all packs\n")
go chkr.Packs(errChan, done)
go chkr.Packs(context.TODO(), errChan)
for err := range errChan {
errorsFound = true
@ -129,7 +127,7 @@ func runCheck(opts CheckOptions, gopts GlobalOptions, args []string) error {
Verbosef("Check snapshots, trees and blobs\n")
errChan = make(chan error)
go chkr.Structure(errChan, done)
go chkr.Structure(context.TODO(), errChan)
for err := range errChan {
errorsFound = true
@ -156,7 +154,7 @@ func runCheck(opts CheckOptions, gopts GlobalOptions, args []string) error {
p := newReadProgress(gopts, restic.Stat{Blobs: chkr.CountPacks()})
errChan := make(chan error)
go chkr.ReadData(p, errChan, done)
go chkr.ReadData(context.TODO(), p, errChan)
for err := range errChan {
errorsFound = true

4
src/cmds/restic/cmd_find.go
View File

@ -187,7 +187,7 @@ func (f *Finder) findInTree(treeID restic.ID, prefix string) error {
debug.Log("%v checking tree %v\n", prefix, treeID.Str())
tree, err := f.repo.LoadTree(treeID)
tree, err := f.repo.LoadTree(context.TODO(), treeID)
if err != nil {
return err
}
@ -283,7 +283,7 @@ func runFind(opts FindOptions, gopts GlobalOptions, args []string) error {
}
}
if err = repo.LoadIndex(); err != nil {
if err = repo.LoadIndex(context.TODO()); err != nil {
return err
}

4
src/cmds/restic/cmd_forget.go
View File

@ -97,7 +97,7 @@ func runForget(opts ForgetOptions, gopts GlobalOptions, args []string) error {
// When explicit snapshots args are given, remove them immediately.
if !opts.DryRun {
h := restic.Handle{Type: restic.SnapshotFile, Name: sn.ID().String()}
if err = repo.Backend().Remove(h); err != nil {
if err = repo.Backend().Remove(context.TODO(), h); err != nil {
return err
}
Verbosef("removed snapshot %v\n", sn.ID().Str())
@ -167,7 +167,7 @@ func runForget(opts ForgetOptions, gopts GlobalOptions, args []string) error {
if !opts.DryRun {
for _, sn := range remove {
h := restic.Handle{Type: restic.SnapshotFile, Name: sn.ID().String()}
err = repo.Backend().Remove(h)
err = repo.Backend().Remove(context.TODO(), h)
if err != nil {
return err
}

3
src/cmds/restic/cmd_init.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"restic/errors"
"restic/repository"
@ -43,7 +44,7 @@ func runInit(gopts GlobalOptions, args []string) error {
s := repository.New(be)
err = s.Init(gopts.password)
err = s.Init(context.TODO(), gopts.password)
if err != nil {
return errors.Fatalf("create key in backend at %s failed: %v\n", gopts.Repo, err)
}

12
src/cmds/restic/cmd_key.go
View File

@ -30,8 +30,8 @@ func listKeys(ctx context.Context, s *repository.Repository) error {
tab.Header = fmt.Sprintf(" %-10s %-10s %-10s %s", "ID", "User", "Host", "Created")
tab.RowFormat = "%s%-10s %-10s %-10s %s"
for id := range s.List(restic.KeyFile, ctx.Done()) {
k, err := repository.LoadKey(s, id.String())
for id := range s.List(ctx, restic.KeyFile) {
k, err := repository.LoadKey(ctx, s, id.String())
if err != nil {
Warnf("LoadKey() failed: %v\n", err)
continue
@ -69,7 +69,7 @@ func addKey(gopts GlobalOptions, repo *repository.Repository) error {
return err
}
id, err := repository.AddKey(repo, pw, repo.Key())
id, err := repository.AddKey(context.TODO(), repo, pw, repo.Key())
if err != nil {
return errors.Fatalf("creating new key failed: %v\n", err)
}
@ -85,7 +85,7 @@ func deleteKey(repo *repository.Repository, name string) error {
}
h := restic.Handle{Type: restic.KeyFile, Name: name}
err := repo.Backend().Remove(h)
err := repo.Backend().Remove(context.TODO(), h)
if err != nil {
return err
}
@ -100,13 +100,13 @@ func changePassword(gopts GlobalOptions, repo *repository.Repository) error {
return err
}
id, err := repository.AddKey(repo, pw, repo.Key())
id, err := repository.AddKey(context.TODO(), repo, pw, repo.Key())
if err != nil {
return errors.Fatalf("creating new key failed: %v\n", err)
}
h := restic.Handle{Type: restic.KeyFile, Name: repo.KeyName()}
err = repo.Backend().Remove(h)
err = repo.Backend().Remove(context.TODO(), h)
if err != nil {
return err
}

5
src/cmds/restic/cmd_list.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"restic"
"restic/errors"
@ -55,7 +56,7 @@ func runList(opts GlobalOptions, args []string) error {
case "locks":
t = restic.LockFile
case "blobs":
idx, err := index.Load(repo, nil)
idx, err := index.Load(context.TODO(), repo, nil)
if err != nil {
return err
}
@ -71,7 +72,7 @@ func runList(opts GlobalOptions, args []string) error {
return errors.Fatal("invalid type")
}
for id := range repo.List(t, nil) {
for id := range repo.List(context.TODO(), t) {
Printf("%s\n", id)
}

4
src/cmds/restic/cmd_ls.go
View File

@ -46,7 +46,7 @@ func init() {
}
func printTree(repo *repository.Repository, id *restic.ID, prefix string) error {
tree, err := repo.LoadTree(*id)
tree, err := repo.LoadTree(context.TODO(), *id)
if err != nil {
return err
}
@ -74,7 +74,7 @@ func runLs(opts LsOptions, gopts GlobalOptions, args []string) error {
return err
}
if err = repo.LoadIndex(); err != nil {
if err = repo.LoadIndex(context.TODO()); err != nil {
return err
}

3
src/cmds/restic/cmd_mount.go
View File

@ -4,6 +4,7 @@
package main
import (
"context"
"os"
"github.com/spf13/cobra"
@ -64,7 +65,7 @@ func mount(opts MountOptions, gopts GlobalOptions, mountpoint string) error {
return err
}
err = repo.LoadIndex()
err = repo.LoadIndex(context.TODO())
if err != nil {
return err
}

20
src/cmds/restic/cmd_prune.go
View File

@ -1,7 +1,6 @@
package main
import (
"context"
"fmt"
"restic"
"restic/debug"
@ -76,14 +75,13 @@ func runPrune(gopts GlobalOptions) error {
}
func pruneRepository(gopts GlobalOptions, repo restic.Repository) error {
err := repo.LoadIndex()
ctx := gopts.ctx
err := repo.LoadIndex(ctx)
if err != nil {
return err
}
ctx, cancel := context.WithCancel(gopts.ctx)
defer cancel()
var stats struct {
blobs int
packs int
@ -92,14 +90,14 @@ func pruneRepository(gopts GlobalOptions, repo restic.Repository) error {
}
Verbosef("counting files in repo\n")
for range repo.List(restic.DataFile, ctx.Done()) {
for range repo.List(ctx, restic.DataFile) {
stats.packs++
}
Verbosef("building new index for repo\n")
bar := newProgressMax(!gopts.Quiet, uint64(stats.packs), "packs")
idx, err := index.New(repo, bar)
idx, err := index.New(ctx, repo, bar)
if err != nil {
return err
}
@ -135,7 +133,7 @@ func pruneRepository(gopts GlobalOptions, repo restic.Repository) error {
Verbosef("load all snapshots\n")
// find referenced blobs
snapshots, err := restic.LoadAllSnapshots(repo)
snapshots, err := restic.LoadAllSnapshots(ctx, repo)
if err != nil {
return err
}
@ -152,7 +150,7 @@ func pruneRepository(gopts GlobalOptions, repo restic.Repository) error {
for _, sn := range snapshots {
debug.Log("process snapshot %v", sn.ID().Str())
err = restic.FindUsedBlobs(repo, *sn.Tree, usedBlobs, seenBlobs)
err = restic.FindUsedBlobs(ctx, repo, *sn.Tree, usedBlobs, seenBlobs)
if err != nil {
return err
}
@ -217,7 +215,7 @@ func pruneRepository(gopts GlobalOptions, repo restic.Repository) error {
if len(rewritePacks) != 0 {
bar = newProgressMax(!gopts.Quiet, uint64(len(rewritePacks)), "packs rewritten")
bar.Start()
err = repository.Repack(repo, rewritePacks, usedBlobs, bar)
err = repository.Repack(ctx, repo, rewritePacks, usedBlobs, bar)
if err != nil {
return err
}
@ -229,7 +227,7 @@ func pruneRepository(gopts GlobalOptions, repo restic.Repository) error {
bar.Start()
for packID := range removePacks {
h := restic.Handle{Type: restic.DataFile, Name: packID.String()}
err = repo.Backend().Remove(h)
err = repo.Backend().Remove(ctx, h)
if err != nil {
Warnf("unable to remove file %v from the repository\n", packID.Str())
}

10
src/cmds/restic/cmd_rebuild_index.go
View File

@ -45,12 +45,12 @@ func rebuildIndex(ctx context.Context, repo restic.Repository) error {
Verbosef("counting files in repo\n")
var packs uint64
for range repo.List(restic.DataFile, ctx.Done()) {
for range repo.List(ctx, restic.DataFile) {
packs++
}
bar := newProgressMax(!globalOptions.Quiet, packs, "packs")
idx, err := index.New(repo, bar)
idx, err := index.New(ctx, repo, bar)
if err != nil {
return err
}
@ -58,11 +58,11 @@ func rebuildIndex(ctx context.Context, repo restic.Repository) error {
Verbosef("finding old index files\n")
var supersedes restic.IDs
for id := range repo.List(restic.IndexFile, ctx.Done()) {
for id := range repo.List(ctx, restic.IndexFile) {
supersedes = append(supersedes, id)
}
id, err := idx.Save(repo, supersedes)
id, err := idx.Save(ctx, repo, supersedes)
if err != nil {
return err
}
@ -72,7 +72,7 @@ func rebuildIndex(ctx context.Context, repo restic.Repository) error {
Verbosef("remove %d old index files\n", len(supersedes))
for _, id := range supersedes {
if err := repo.Backend().Remove(restic.Handle{
if err := repo.Backend().Remove(ctx, restic.Handle{
Type: restic.IndexFile,
Name: id.String(),
}); err != nil {

8
src/cmds/restic/cmd_restore.go
View File

@ -50,6 +50,8 @@ func init() {
}
func runRestore(opts RestoreOptions, gopts GlobalOptions, args []string) error {
ctx := gopts.ctx
if len(args) != 1 {
return errors.Fatal("no snapshot ID specified")
}
@ -79,7 +81,7 @@ func runRestore(opts RestoreOptions, gopts GlobalOptions, args []string) error {
}
}
err = repo.LoadIndex()
err = repo.LoadIndex(ctx)
if err != nil {
return err
}
@ -87,7 +89,7 @@ func runRestore(opts RestoreOptions, gopts GlobalOptions, args []string) error {
var id restic.ID
if snapshotIDString == "latest" {
id, err = restic.FindLatestSnapshot(repo, opts.Paths, opts.Tags, opts.Host)
id, err = restic.FindLatestSnapshot(ctx, repo, opts.Paths, opts.Tags, opts.Host)
if err != nil {
Exitf(1, "latest snapshot for criteria not found: %v Paths:%v Host:%v", err, opts.Paths, opts.Host)
}
@ -136,7 +138,7 @@ func runRestore(opts RestoreOptions, gopts GlobalOptions, args []string) error {
Verbosef("restoring %s to %s\n", res.Snapshot(), opts.Target)
err = res.RestoreTo(opts.Target)
err = res.RestoreTo(ctx, opts.Target)
if totalErrors > 0 {
Printf("There were %d errors\n", totalErrors)
}

4
src/cmds/restic/cmd_tag.go
View File

@ -76,7 +76,7 @@ func changeTags(repo *repository.Repository, sn *restic.Snapshot, setTags, addTa
}
// Save the new snapshot.
id, err := repo.SaveJSONUnpacked(restic.SnapshotFile, sn)
id, err := repo.SaveJSONUnpacked(context.TODO(), restic.SnapshotFile, sn)
if err != nil {
return false, err
}
@ -89,7 +89,7 @@ func changeTags(repo *repository.Repository, sn *restic.Snapshot, setTags, addTa
// Remove the old snapshot.
h := restic.Handle{Type: restic.SnapshotFile, Name: sn.ID().String()}
if err = repo.Backend().Remove(h); err != nil {
if err = repo.Backend().Remove(context.TODO(), h); err != nil {
return false, err
}

3
src/cmds/restic/cmd_unlock.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"restic"
"github.com/spf13/cobra"
@ -41,7 +42,7 @@ func runUnlock(opts UnlockOptions, gopts GlobalOptions) error {
fn = restic.RemoveAllLocks
}
err = fn(repo)
err = fn(context.TODO(), repo)
if err != nil {
return err
}

8
src/cmds/restic/find.go
View File

@ -22,7 +22,7 @@ func FindFilteredSnapshots(ctx context.Context, repo *repository.Repository, hos
// Process all snapshot IDs given as arguments.
for _, s := range snapshotIDs {
if s == "latest" {
id, err = restic.FindLatestSnapshot(repo, paths, tags, host)
id, err = restic.FindLatestSnapshot(ctx, repo, paths, tags, host)
if err != nil {
Warnf("Ignoring %q, no snapshot matched given filter (Paths:%v Tags:%v Host:%v)\n", s, paths, tags, host)
usedFilter = true
@ -44,7 +44,7 @@ func FindFilteredSnapshots(ctx context.Context, repo *repository.Repository, hos
}
for _, id := range ids.Uniq() {
sn, err := restic.LoadSnapshot(repo, id)
sn, err := restic.LoadSnapshot(ctx, repo, id)
if err != nil {
Warnf("Ignoring %q, could not load snapshot: %v\n", id, err)
continue
@ -58,8 +58,8 @@ func FindFilteredSnapshots(ctx context.Context, repo *repository.Repository, hos
return
}
for id := range repo.List(restic.SnapshotFile, ctx.Done()) {
sn, err := restic.LoadSnapshot(repo, id)
for id := range repo.List(ctx, restic.SnapshotFile) {
sn, err := restic.LoadSnapshot(ctx, repo, id)
if err != nil {
Warnf("Ignoring %q, could not load snapshot: %v\n", id, err)
continue

4
src/cmds/restic/global.go
View File

@ -310,7 +310,7 @@ func OpenRepository(opts GlobalOptions) (*repository.Repository, error) {
}
}
err = s.SearchKey(opts.password, maxKeys)
err = s.SearchKey(context.TODO(), opts.password, maxKeys)
if err != nil {
return nil, errors.Fatalf("unable to open repo: %v", err)
}
@ -440,7 +440,7 @@ func open(s string, opts options.Options) (restic.Backend, error) {
}
// check if config is there
fi, err := be.Stat(restic.Handle{Type: restic.ConfigFile})
fi, err := be.Stat(context.TODO(), restic.Handle{Type: restic.ConfigFile})
if err != nil {
return nil, errors.Fatalf("unable to open config file: %v\nIs there a repository at the following location?\n%v", err, s)
}

5
src/cmds/restic/lock.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"os"
"sync"
@ -32,7 +33,7 @@ func lockRepository(repo *repository.Repository, exclusive bool) (*restic.Lock,
lockFn = restic.NewExclusiveLock
}
lock, err := lockFn(repo)
lock, err := lockFn(context.TODO(), repo)
if err != nil {
return nil, err
}
@ -75,7 +76,7 @@ func refreshLocks(wg *sync.WaitGroup, done <-chan struct{}) {
debug.Log("refreshing locks")
globalLocks.Lock()
for _, lock := range globalLocks.locks {
err := lock.Refresh()
err := lock.Refresh(context.TODO())
if err != nil {
fmt.Fprintf(os.Stderr, "unable to refresh lock: %v\n", err)
}

11
src/restic/archiver/archive_reader.go
View File

@ -1,6 +1,7 @@
package archiver
import (
"context"
"io"
"restic"
"restic/debug"
@ -20,7 +21,7 @@ type Reader struct {
}
// Archive reads data from the reader and saves it to the repo.
func (r *Reader) Archive(name string, rd io.Reader, p *restic.Progress) (*restic.Snapshot, restic.ID, error) {
func (r *Reader) Archive(ctx context.Context, name string, rd io.Reader, p *restic.Progress) (*restic.Snapshot, restic.ID, error) {
if name == "" {
return nil, restic.ID{}, errors.New("no filename given")
}
@ -53,7 +54,7 @@ func (r *Reader) Archive(name string, rd io.Reader, p *restic.Progress) (*restic
id := restic.Hash(chunk.Data)
if !repo.Index().Has(id, restic.DataBlob) {
_, err := repo.SaveBlob(restic.DataBlob, chunk.Data, id)
_, err := repo.SaveBlob(ctx, restic.DataBlob, chunk.Data, id)
if err != nil {
return nil, restic.ID{}, err
}
@ -87,14 +88,14 @@ func (r *Reader) Archive(name string, rd io.Reader, p *restic.Progress) (*restic
},
}
treeID, err := repo.SaveTree(tree)
treeID, err := repo.SaveTree(ctx, tree)
if err != nil {
return nil, restic.ID{}, err
}
sn.Tree = &treeID
debug.Log("tree saved as %v", treeID.Str())
id, err := repo.SaveJSONUnpacked(restic.SnapshotFile, sn)
id, err := repo.SaveJSONUnpacked(ctx, restic.SnapshotFile, sn)
if err != nil {
return nil, restic.ID{}, err
}
@ -106,7 +107,7 @@ func (r *Reader) Archive(name string, rd io.Reader, p *restic.Progress) (*restic
return nil, restic.ID{}, err
}
err = repo.SaveIndex()
err = repo.SaveIndex(ctx)
if err != nil {
return nil, restic.ID{}, err
}

70
src/restic/archiver/archiver.go
View File

@ -1,6 +1,7 @@
package archiver
import (
"context"
"encoding/json"
"fmt"
"io"
@ -92,7 +93,7 @@ func (arch *Archiver) isKnownBlob(id restic.ID, t restic.BlobType) bool {
}
// Save stores a blob read from rd in the repository.
func (arch *Archiver) Save(t restic.BlobType, data []byte, id restic.ID) error {
func (arch *Archiver) Save(ctx context.Context, t restic.BlobType, data []byte, id restic.ID) error {
debug.Log("Save(%v, %v)\n", t, id.Str())
if arch.isKnownBlob(id, restic.DataBlob) {
@ -100,7 +101,7 @@ func (arch *Archiver) Save(t restic.BlobType, data []byte, id restic.ID) error {
return nil
}
_, err := arch.repo.SaveBlob(t, data, id)
_, err := arch.repo.SaveBlob(ctx, t, data, id)
if err != nil {
debug.Log("Save(%v, %v): error %v\n", t, id.Str(), err)
return err
@ -111,7 +112,7 @@ func (arch *Archiver) Save(t restic.BlobType, data []byte, id restic.ID) error {
}
// SaveTreeJSON stores a tree in the repository.
func (arch *Archiver) SaveTreeJSON(tree *restic.Tree) (restic.ID, error) {
func (arch *Archiver) SaveTreeJSON(ctx context.Context, tree *restic.Tree) (restic.ID, error) {
data, err := json.Marshal(tree)
if err != nil {
return restic.ID{}, errors.Wrap(err, "Marshal")
@ -124,7 +125,7 @@ func (arch *Archiver) SaveTreeJSON(tree *restic.Tree) (restic.ID, error) {
return id, nil
}
return arch.repo.SaveBlob(restic.TreeBlob, data, id)
return arch.repo.SaveBlob(ctx, restic.TreeBlob, data, id)
}
func (arch *Archiver) reloadFileIfChanged(node *restic.Node, file fs.File) (*restic.Node, error) {
@ -153,11 +154,11 @@ type saveResult struct {
bytes uint64
}
func (arch *Archiver) saveChunk(chunk chunker.Chunk, p *restic.Progress, token struct{}, file fs.File, resultChannel chan<- saveResult) {
func (arch *Archiver) saveChunk(ctx context.Context, chunk chunker.Chunk, p *restic.Progress, token struct{}, file fs.File, resultChannel chan<- saveResult) {
defer freeBuf(chunk.Data)
id := restic.Hash(chunk.Data)
err := arch.Save(restic.DataBlob, chunk.Data, id)
err := arch.Save(ctx, restic.DataBlob, chunk.Data, id)
// TODO handle error
if err != nil {
panic(err)
@ -206,7 +207,7 @@ func updateNodeContent(node *restic.Node, results []saveResult) error {
// SaveFile stores the content of the file on the backend as a Blob by calling
// Save for each chunk.
func (arch *Archiver) SaveFile(p *restic.Progress, node *restic.Node) (*restic.Node, error) {
func (arch *Archiver) SaveFile(ctx context.Context, p *restic.Progress, node *restic.Node) (*restic.Node, error) {
file, err := fs.Open(node.Path)
defer file.Close()
if err != nil {
@ -234,7 +235,7 @@ func (arch *Archiver) SaveFile(p *restic.Progress, node *restic.Node) (*restic.N
}
resCh := make(chan saveResult, 1)
go arch.saveChunk(chunk, p, <-arch.blobToken, file, resCh)
go arch.saveChunk(ctx, chunk, p, <-arch.blobToken, file, resCh)
resultChannels = append(resultChannels, resCh)
}
@ -247,7 +248,7 @@ func (arch *Archiver) SaveFile(p *restic.Progress, node *restic.Node) (*restic.N
return node, err
}
func (arch *Archiver) fileWorker(wg *sync.WaitGroup, p *restic.Progress, done <-chan struct{}, entCh <-chan pipe.Entry) {
func (arch *Archiver) fileWorker(ctx context.Context, wg *sync.WaitGroup, p *restic.Progress, entCh <-chan pipe.Entry) {
defer func() {
debug.Log("done")
wg.Done()
@ -305,7 +306,7 @@ func (arch *Archiver) fileWorker(wg *sync.WaitGroup, p *restic.Progress, done <-
// otherwise read file normally
if node.Type == "file" && len(node.Content) == 0 {
debug.Log(" read and save %v", e.Path())
node, err = arch.SaveFile(p, node)
node, err = arch.SaveFile(ctx, p, node)
if err != nil {
fmt.Fprintf(os.Stderr, "error for %v: %v\n", node.Path, err)
arch.Warn(e.Path(), nil, err)
@ -322,14 +323,14 @@ func (arch *Archiver) fileWorker(wg *sync.WaitGroup, p *restic.Progress, done <-
debug.Log(" processed %v, %d blobs", e.Path(), len(node.Content))
e.Result() <- node
p.Report(restic.Stat{Files: 1})
case <-done:
case <-ctx.Done():
// pipeline was cancelled
return
}
}
}
func (arch *Archiver) dirWorker(wg *sync.WaitGroup, p *restic.Progress, done <-chan struct{}, dirCh <-chan pipe.Dir) {
func (arch *Archiver) dirWorker(ctx context.Context, wg *sync.WaitGroup, p *restic.Progress, dirCh <-chan pipe.Dir) {
debug.Log("start")
defer func() {
debug.Log("done")
@ -398,7 +399,7 @@ func (arch *Archiver) dirWorker(wg *sync.WaitGroup, p *restic.Progress, done <-c
node.Error = err.Error()
}
id, err := arch.SaveTreeJSON(tree)
id, err := arch.SaveTreeJSON(ctx, tree)
if err != nil {
panic(err)
}
@ -415,7 +416,7 @@ func (arch *Archiver) dirWorker(wg *sync.WaitGroup, p *restic.Progress, done <-c
if dir.Path() != "" {
p.Report(restic.Stat{Dirs: 1})
}
case <-done:
case <-ctx.Done():
// pipeline was cancelled
return
}
@ -427,7 +428,7 @@ type archivePipe struct {
New <-chan pipe.Job
}
func copyJobs(done <-chan struct{}, in <-chan pipe.Job, out chan<- pipe.Job) {
func copyJobs(ctx context.Context, in <-chan pipe.Job, out chan<- pipe.Job) {
var (
// disable sending on the outCh until we received a job
outCh chan<- pipe.Job
@ -439,7 +440,7 @@ func copyJobs(done <-chan struct{}, in <-chan pipe.Job, out chan<- pipe.Job) {
for {
select {
case <-done:
case <-ctx.Done():
return
case job, ok = <-inCh:
if !ok {
@ -462,7 +463,7 @@ type archiveJob struct {
new pipe.Job
}
func (a *archivePipe) compare(done <-chan struct{}, out chan<- pipe.Job) {
func (a *archivePipe) compare(ctx context.Context, out chan<- pipe.Job) {
defer func() {
close(out)
debug.Log("done")
@ -488,7 +489,7 @@ func (a *archivePipe) compare(done <-chan struct{}, out chan<- pipe.Job) {
out <- archiveJob{new: newJob}.Copy()
}
copyJobs(done, a.New, out)
copyJobs(ctx, a.New, out)
return
}
@ -585,7 +586,7 @@ func (j archiveJob) Copy() pipe.Job {
const saveIndexTime = 30 * time.Second
// saveIndexes regularly queries the master index for full indexes and saves them.
func (arch *Archiver) saveIndexes(wg *sync.WaitGroup, done <-chan struct{}) {
func (arch *Archiver) saveIndexes(ctx context.Context, wg *sync.WaitGroup) {
defer wg.Done()
ticker := time.NewTicker(saveIndexTime)
@ -593,11 +594,11 @@ func (arch *Archiver) saveIndexes(wg *sync.WaitGroup, done <-chan struct{}) {
for {
select {
case <-done:
case <-ctx.Done():
return
case <-ticker.C:
debug.Log("saving full indexes")
err := arch.repo.SaveFullIndex()
err := arch.repo.SaveFullIndex(ctx)
if err != nil {
debug.Log("save indexes returned an error: %v", err)
fmt.Fprintf(os.Stderr, "error saving preliminary index: %v\n", err)
@ -634,7 +635,7 @@ func (p baseNameSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
// Snapshot creates a snapshot of the given paths. If parentrestic.ID is set, this is
// used to compare the files to the ones archived at the time this snapshot was
// taken.
func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostname string, parentID *restic.ID) (*restic.Snapshot, restic.ID, error) {
func (arch *Archiver) Snapshot(ctx context.Context, p *restic.Progress, paths, tags []string, hostname string, parentID *restic.ID) (*restic.Snapshot, restic.ID, error) {
paths = unique(paths)
sort.Sort(baseNameSlice(paths))
@ -643,7 +644,6 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
debug.RunHook("Archiver.Snapshot", nil)
// signal the whole pipeline to stop
done := make(chan struct{})
var err error
p.Start()
@ -663,14 +663,14 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
sn.Parent = parentID
// load parent snapshot
parent, err := restic.LoadSnapshot(arch.repo, *parentID)
parent, err := restic.LoadSnapshot(ctx, arch.repo, *parentID)
if err != nil {
return nil, restic.ID{}, err
}
// start walker on old tree
ch := make(chan walk.TreeJob)
go walk.Tree(arch.repo, *parent.Tree, done, ch)
go walk.Tree(ctx, arch.repo, *parent.Tree, ch)
jobs.Old = ch
} else {
// use closed channel
@ -683,13 +683,13 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
pipeCh := make(chan pipe.Job)
resCh := make(chan pipe.Result, 1)
go func() {
pipe.Walk(paths, arch.SelectFilter, done, pipeCh, resCh)
pipe.Walk(ctx, paths, arch.SelectFilter, pipeCh, resCh)
debug.Log("pipe.Walk done")
}()
jobs.New = pipeCh
ch := make(chan pipe.Job)
go jobs.compare(done, ch)
go jobs.compare(ctx, ch)
var wg sync.WaitGroup
entCh := make(chan pipe.Entry)
@ -708,22 +708,22 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
// run workers
for i := 0; i < maxConcurrency; i++ {
wg.Add(2)
go arch.fileWorker(&wg, p, done, entCh)
go arch.dirWorker(&wg, p, done, dirCh)
go arch.fileWorker(ctx, &wg, p, entCh)
go arch.dirWorker(ctx, &wg, p, dirCh)
}
// run index saver
var wgIndexSaver sync.WaitGroup
stopIndexSaver := make(chan struct{})
indexCtx, indexCancel := context.WithCancel(ctx)
wgIndexSaver.Add(1)
go arch.saveIndexes(&wgIndexSaver, stopIndexSaver)
go arch.saveIndexes(indexCtx, &wgIndexSaver)
// wait for all workers to terminate
debug.Log("wait for workers")
wg.Wait()
// stop index saver
close(stopIndexSaver)
indexCancel()
wgIndexSaver.Wait()
debug.Log("workers terminated")
@ -740,7 +740,7 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
sn.Tree = root.Subtree
// load top-level tree again to see if it is empty
toptree, err := arch.repo.LoadTree(*root.Subtree)
toptree, err := arch.repo.LoadTree(ctx, *root.Subtree)
if err != nil {
return nil, restic.ID{}, err
}
@ -750,7 +750,7 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
}
// save index
err = arch.repo.SaveIndex()
err = arch.repo.SaveIndex(ctx)
if err != nil {
debug.Log("error saving index: %v", err)
return nil, restic.ID{}, err
@ -759,7 +759,7 @@ func (arch *Archiver) Snapshot(p *restic.Progress, paths, tags []string, hostnam
debug.Log("saved indexes")
// save snapshot
id, err := arch.repo.SaveJSONUnpacked(restic.SnapshotFile, sn)
id, err := arch.repo.SaveJSONUnpacked(ctx, restic.SnapshotFile, sn)
if err != nil {
return nil, restic.ID{}, err
}

3
src/restic/archiver/testing.go
View File

@ -1,6 +1,7 @@
package archiver
import (
"context"
"restic"
"testing"
)
@ -8,7 +9,7 @@ import (
// TestSnapshot creates a new snapshot of path.
func TestSnapshot(t testing.TB, repo restic.Repository, path string, parent *restic.ID) *restic.Snapshot {
arch := New(repo)
sn, _, err := arch.Snapshot(nil, []string{path}, []string{"test"}, "localhost", parent)
sn, _, err := arch.Snapshot(context.TODO(), nil, []string{path}, []string{"test"}, "localhost", parent)
if err != nil {
t.Fatal(err)
}

81
src/restic/checker/checker.go
View File

@ -1,6 +1,7 @@
package checker
import (
"context"
"crypto/sha256"
"fmt"
"io"
@ -76,7 +77,7 @@ func (err ErrOldIndexFormat) Error() string {
}
// LoadIndex loads all index files.
func (c *Checker) LoadIndex() (hints []error, errs []error) {
func (c *Checker) LoadIndex(ctx context.Context) (hints []error, errs []error) {
debug.Log("Start")
type indexRes struct {
Index *repository.Index
@ -86,21 +87,21 @@ func (c *Checker) LoadIndex() (hints []error, errs []error) {
indexCh := make(chan indexRes)
worker := func(id restic.ID, done <-chan struct{}) error {
worker := func(ctx context.Context, id restic.ID) error {
debug.Log("worker got index %v", id)
idx, err := repository.LoadIndexWithDecoder(c.repo, id, repository.DecodeIndex)
idx, err := repository.LoadIndexWithDecoder(ctx, c.repo, id, repository.DecodeIndex)
if errors.Cause(err) == repository.ErrOldIndexFormat {
debug.Log("index %v has old format", id.Str())
hints = append(hints, ErrOldIndexFormat{id})
idx, err = repository.LoadIndexWithDecoder(c.repo, id, repository.DecodeOldIndex)
idx, err = repository.LoadIndexWithDecoder(ctx, c.repo, id, repository.DecodeOldIndex)
}
err = errors.Wrapf(err, "error loading index %v", id.Str())
select {
case indexCh <- indexRes{Index: idx, ID: id.String(), err: err}:
case <-done:
case <-ctx.Done():
}
return nil
@ -109,7 +110,7 @@ func (c *Checker) LoadIndex() (hints []error, errs []error) {
go func() {
defer close(indexCh)
debug.Log("start loading indexes in parallel")
err := repository.FilesInParallel(c.repo.Backend(), restic.IndexFile, defaultParallelism,
err := repository.FilesInParallel(ctx, c.repo.Backend(), restic.IndexFile, defaultParallelism,
repository.ParallelWorkFuncParseID(worker))
debug.Log("loading indexes finished, error: %v", err)
if err != nil {
@ -183,7 +184,7 @@ func (e PackError) Error() string {
return "pack " + e.ID.String() + ": " + e.Err.Error()
}
func packIDTester(repo restic.Repository, inChan <-chan restic.ID, errChan chan<- error, wg *sync.WaitGroup, done <-chan struct{}) {
func packIDTester(ctx context.Context, repo restic.Repository, inChan <-chan restic.ID, errChan chan<- error, wg *sync.WaitGroup) {
debug.Log("worker start")
defer debug.Log("worker done")
@ -191,7 +192,7 @@ func packIDTester(repo restic.Repository, inChan <-chan restic.ID, errChan chan<
for id := range inChan {
h := restic.Handle{Type: restic.DataFile, Name: id.String()}
ok, err := repo.Backend().Test(h)
ok, err := repo.Backend().Test(ctx, h)
if err != nil {
err = PackError{ID: id, Err: err}
} else {
@ -203,7 +204,7 @@ func packIDTester(repo restic.Repository, inChan <-chan restic.ID, errChan chan<
if err != nil {
debug.Log("error checking for pack %s: %v", id.Str(), err)
select {
case <-done:
case <-ctx.Done():
return
case errChan <- err:
}
@ -218,7 +219,7 @@ func packIDTester(repo restic.Repository, inChan <-chan restic.ID, errChan chan<
// Packs checks that all packs referenced in the index are still available and
// there are no packs that aren't in an index. errChan is closed after all
// packs have been checked.
func (c *Checker) Packs(errChan chan<- error, done <-chan struct{}) {
func (c *Checker) Packs(ctx context.Context, errChan chan<- error) {
defer close(errChan)
debug.Log("checking for %d packs", len(c.packs))
@ -229,7 +230,7 @@ func (c *Checker) Packs(errChan chan<- error, done <-chan struct{}) {
IDChan := make(chan restic.ID)
for i := 0; i < defaultParallelism; i++ {
workerWG.Add(1)
go packIDTester(c.repo, IDChan, errChan, &workerWG, done)
go packIDTester(ctx, c.repo, IDChan, errChan, &workerWG)
}
for id := range c.packs {
@ -242,12 +243,12 @@ func (c *Checker) Packs(errChan chan<- error, done <-chan struct{}) {
workerWG.Wait()
debug.Log("workers terminated")
for id := range c.repo.List(restic.DataFile, done) {
for id := range c.repo.List(ctx, restic.DataFile) {
debug.Log("check data blob %v", id.Str())
if !seenPacks.Has(id) {
c.orphanedPacks = append(c.orphanedPacks, id)
select {
case <-done:
case <-ctx.Done():
return
case errChan <- PackError{ID: id, Orphaned: true, Err: errors.New("not referenced in any index")}:
}
@ -277,8 +278,8 @@ func (e Error) Error() string {
return e.Err.Error()
}
func loadTreeFromSnapshot(repo restic.Repository, id restic.ID) (restic.ID, error) {
sn, err := restic.LoadSnapshot(repo, id)
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.Str(), err)
return restic.ID{}, err
@ -293,7 +294,7 @@ func loadTreeFromSnapshot(repo restic.Repository, id restic.ID) (restic.ID, erro
}
// loadSnapshotTreeIDs loads all snapshots from backend and returns the tree IDs.
func loadSnapshotTreeIDs(repo restic.Repository) (restic.IDs, []error) {
func loadSnapshotTreeIDs(ctx context.Context, repo restic.Repository) (restic.IDs, []error) {
var trees struct {
IDs restic.IDs
sync.Mutex
@ -304,7 +305,7 @@ func loadSnapshotTreeIDs(repo restic.Repository) (restic.IDs, []error) {
sync.Mutex
}
snapshotWorker := func(strID string, done <-chan struct{}) error {
snapshotWorker := func(ctx context.Context, strID string) error {
id, err := restic.ParseID(strID)
if err != nil {
return err
@ -312,7 +313,7 @@ func loadSnapshotTreeIDs(repo restic.Repository) (restic.IDs, []error) {
debug.Log("load snapshot %v", id.Str())
treeID, err := loadTreeFromSnapshot(repo, id)
treeID, err := loadTreeFromSnapshot(ctx, repo, id)
if err != nil {
errs.Lock()
errs.errs = append(errs.errs, err)
@ -328,7 +329,7 @@ func loadSnapshotTreeIDs(repo restic.Repository) (restic.IDs, []error) {
return nil
}
err := repository.FilesInParallel(repo.Backend(), restic.SnapshotFile, defaultParallelism, snapshotWorker)
err := repository.FilesInParallel(ctx, repo.Backend(), restic.SnapshotFile, defaultParallelism, snapshotWorker)
if err != nil {
errs.errs = append(errs.errs, err)
}
@ -353,9 +354,9 @@ type treeJob struct {
}
// loadTreeWorker loads trees from repo and sends them to out.
func loadTreeWorker(repo restic.Repository,
func loadTreeWorker(ctx context.Context, repo restic.Repository,
in <-chan restic.ID, out chan<- treeJob,
done <-chan struct{}, wg *sync.WaitGroup) {
wg *sync.WaitGroup) {
defer func() {
debug.Log("exiting")
@ -371,7 +372,7 @@ func loadTreeWorker(repo restic.Repository,
outCh = nil
for {
select {
case <-done:
case <-ctx.Done():
return
case treeID, ok := <-inCh:
@ -380,7 +381,7 @@ func loadTreeWorker(repo restic.Repository,
}
debug.Log("load tree %v", treeID.Str())
tree, err := repo.LoadTree(treeID)
tree, err := repo.LoadTree(ctx, treeID)
debug.Log("load tree %v (%v) returned err: %v", tree, treeID.Str(), err)
job = treeJob{ID: treeID, error: err, Tree: tree}
outCh = out
@ -395,7 +396,7 @@ func loadTreeWorker(repo restic.Repository,
}
// checkTreeWorker checks the trees received and sends out errors to errChan.
func (c *Checker) checkTreeWorker(in <-chan treeJob, out chan<- error, done <-chan struct{}, wg *sync.WaitGroup) {
func (c *Checker) checkTreeWorker(ctx context.Context, in <-chan treeJob, out chan<- error, wg *sync.WaitGroup) {
defer func() {
debug.Log("exiting")
wg.Done()
@ -410,7 +411,7 @@ func (c *Checker) checkTreeWorker(in <-chan treeJob, out chan<- error, done <-ch
outCh = nil
for {
select {
case <-done:
case <-ctx.Done():
debug.Log("done channel closed, exiting")
return
@ -458,7 +459,7 @@ func (c *Checker) checkTreeWorker(in <-chan treeJob, out chan<- error, done <-ch
}
}
func filterTrees(backlog restic.IDs, loaderChan chan<- restic.ID, in <-chan treeJob, out chan<- treeJob, done <-chan struct{}) {
func filterTrees(ctx context.Context, backlog restic.IDs, loaderChan chan<- restic.ID, in <-chan treeJob, out chan<- treeJob) {
defer func() {
debug.Log("closing output channels")
close(loaderChan)
@ -489,7 +490,7 @@ func filterTrees(backlog restic.IDs, loaderChan chan<- restic.ID, in <-chan tree
}
select {
case <-done:
case <-ctx.Done():
return
case loadCh <- nextTreeID:
@ -549,15 +550,15 @@ func filterTrees(backlog restic.IDs, loaderChan chan<- restic.ID, in <-chan tree
// 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.
func (c *Checker) Structure(errChan chan<- error, done <-chan struct{}) {
func (c *Checker) Structure(ctx context.Context, errChan chan<- error) {
defer close(errChan)
trees, errs := loadSnapshotTreeIDs(c.repo)
trees, errs := loadSnapshotTreeIDs(ctx, c.repo)
debug.Log("need to check %d trees from snapshots, %d errs returned", len(trees), len(errs))
for _, err := range errs {
select {
case <-done:
case <-ctx.Done():
return
case errChan <- err:
}
@ -570,11 +571,11 @@ func (c *Checker) Structure(errChan chan<- error, done <-chan struct{}) {
var wg sync.WaitGroup
for i := 0; i < defaultParallelism; i++ {
wg.Add(2)
go loadTreeWorker(c.repo, treeIDChan, treeJobChan1, done, &wg)
go c.checkTreeWorker(treeJobChan2, errChan, done, &wg)
go loadTreeWorker(ctx, c.repo, treeIDChan, treeJobChan1, &wg)
go c.checkTreeWorker(ctx, treeJobChan2, errChan, &wg)
}
filterTrees(trees, treeIDChan, treeJobChan1, treeJobChan2, done)
filterTrees(ctx, trees, treeIDChan, treeJobChan1, treeJobChan2)
wg.Wait()
}
@ -659,11 +660,11 @@ func (c *Checker) CountPacks() uint64 {
}
// checkPack reads a pack and checks the integrity of all blobs.
func checkPack(r restic.Repository, id restic.ID) error {
func checkPack(ctx context.Context, r restic.Repository, id restic.ID) error {
debug.Log("checking pack %v", id.Str())
h := restic.Handle{Type: restic.DataFile, Name: id.String()}
rd, err := r.Backend().Load(h, 0, 0)
rd, err := r.Backend().Load(ctx, h, 0, 0)
if err != nil {
return err
}
@ -748,7 +749,7 @@ func checkPack(r restic.Repository, id restic.ID) error {
}
// ReadData loads all data from the repository and checks the integrity.
func (c *Checker) ReadData(p *restic.Progress, errChan chan<- error, done <-chan struct{}) {
func (c *Checker) ReadData(ctx context.Context, p *restic.Progress, errChan chan<- error) {
defer close(errChan)
p.Start()
@ -761,7 +762,7 @@ func (c *Checker) ReadData(p *restic.Progress, errChan chan<- error, done <-chan
var ok bool
select {
case <-done:
case <-ctx.Done():
return
case id, ok = <-in:
if !ok {
@ -769,21 +770,21 @@ func (c *Checker) ReadData(p *restic.Progress, errChan chan<- error, done <-chan
}
}
err := checkPack(c.repo, id)
err := checkPack(ctx, c.repo, id)
p.Report(restic.Stat{Blobs: 1})
if err == nil {
continue
}
select {
case <-done:
case <-ctx.Done():
return
case errChan <- err:
}
}
}
ch := c.repo.List(restic.DataFile, done)
ch := c.repo.List(ctx, restic.DataFile)
var wg sync.WaitGroup
for i := 0; i < defaultParallelism; i++ {

12
src/restic/checker/testing.go
View File

@ -1,6 +1,7 @@
package checker
import (
"context"
"restic"
"testing"
)
@ -9,7 +10,7 @@ import (
func TestCheckRepo(t testing.TB, repo restic.Repository) {
chkr := New(repo)
hints, errs := chkr.LoadIndex()
hints, errs := chkr.LoadIndex(context.TODO())
if len(errs) != 0 {
t.Fatalf("errors loading index: %v", errs)
}
@ -18,12 +19,9 @@ func TestCheckRepo(t testing.TB, repo restic.Repository) {
t.Fatalf("errors loading index: %v", hints)
}
done := make(chan struct{})
defer close(done)
// packs
errChan := make(chan error)
go chkr.Packs(errChan, done)
go chkr.Packs(context.TODO(), errChan)
for err := range errChan {
t.Error(err)
@ -31,7 +29,7 @@ func TestCheckRepo(t testing.TB, repo restic.Repository) {
// structure
errChan = make(chan error)
go chkr.Structure(errChan, done)
go chkr.Structure(context.TODO(), errChan)
for err := range errChan {
t.Error(err)
@ -45,7 +43,7 @@ func TestCheckRepo(t testing.TB, repo restic.Repository) {
// read data
errChan = make(chan error)
go chkr.ReadData(nil, errChan, done)
go chkr.ReadData(context.TODO(), nil, errChan)
for err := range errChan {
t.Error(err)

7
src/restic/config.go
View File

@ -1,6 +1,7 @@
package restic
import (
"context"
"testing"
"restic/errors"
@ -23,7 +24,7 @@ const RepoVersion = 1
// JSONUnpackedLoader loads unpacked JSON.
type JSONUnpackedLoader interface {
LoadJSONUnpacked(FileType, ID, interface{}) error
LoadJSONUnpacked(context.Context, FileType, ID, interface{}) error
}
// CreateConfig creates a config file with a randomly selected polynomial and
@ -57,12 +58,12 @@ func TestCreateConfig(t testing.TB, pol chunker.Pol) (cfg Config) {
}
// LoadConfig returns loads, checks and returns the config for a repository.
func LoadConfig(r JSONUnpackedLoader) (Config, error) {
func LoadConfig(ctx context.Context, r JSONUnpackedLoader) (Config, error) {
var (
cfg Config
)
err := r.LoadJSONUnpacked(ConfigFile, ID{}, &cfg)
err := r.LoadJSONUnpacked(ctx, ConfigFile, ID{}, &cfg)
if err != nil {
return Config{}, err
}

8
src/restic/find.go
View File

@ -1,12 +1,14 @@
package restic
import "context"
// FindUsedBlobs traverses the tree ID and adds all seen blobs (trees and data
// blobs) to the set blobs. The tree blobs in the `seen` BlobSet will not be visited
// again.
func FindUsedBlobs(repo Repository, treeID ID, blobs BlobSet, seen BlobSet) error {
func FindUsedBlobs(ctx context.Context, repo Repository, treeID ID, blobs BlobSet, seen BlobSet) error {
blobs.Insert(BlobHandle{ID: treeID, Type: TreeBlob})
tree, err := repo.LoadTree(treeID)
tree, err := repo.LoadTree(ctx, treeID)
if err != nil {
return err
}
@ -26,7 +28,7 @@ func FindUsedBlobs(repo Repository, treeID ID, blobs BlobSet, seen BlobSet) erro
seen.Insert(h)
err := FindUsedBlobs(repo, subtreeID, blobs, seen)
err := FindUsedBlobs(ctx, repo, subtreeID, blobs, seen)
if err != nil {
return err
}

16
src/restic/fuse/dir.go
View File

@ -26,9 +26,9 @@ type dir struct {
ownerIsRoot bool
}
func newDir(repo restic.Repository, node *restic.Node, ownerIsRoot bool) (*dir, error) {
func newDir(ctx context.Context, repo restic.Repository, node *restic.Node, ownerIsRoot bool) (*dir, error) {
debug.Log("new dir for %v (%v)", node.Name, node.Subtree.Str())
tree, err := repo.LoadTree(*node.Subtree)
tree, err := repo.LoadTree(ctx, *node.Subtree)
if err != nil {
debug.Log(" error loading tree %v: %v", node.Subtree.Str(), err)
return nil, err
@ -49,7 +49,7 @@ func newDir(repo restic.Repository, node *restic.Node, ownerIsRoot bool) (*dir,
// replaceSpecialNodes replaces nodes with name "." and "/" by their contents.
// Otherwise, the node is returned.
func replaceSpecialNodes(repo restic.Repository, node *restic.Node) ([]*restic.Node, error) {
func replaceSpecialNodes(ctx context.Context, repo restic.Repository, node *restic.Node) ([]*restic.Node, error) {
if node.Type != "dir" || node.Subtree == nil {
return []*restic.Node{node}, nil
}
@ -58,7 +58,7 @@ func replaceSpecialNodes(repo restic.Repository, node *restic.Node) ([]*restic.N
return []*restic.Node{node}, nil
}
tree, err := repo.LoadTree(*node.Subtree)
tree, err := repo.LoadTree(ctx, *node.Subtree)
if err != nil {
return nil, err
}
@ -66,16 +66,16 @@ func replaceSpecialNodes(repo restic.Repository, node *restic.Node) ([]*restic.N
return tree.Nodes, nil
}
func newDirFromSnapshot(repo restic.Repository, snapshot SnapshotWithId, ownerIsRoot bool) (*dir, error) {
func newDirFromSnapshot(ctx context.Context, repo restic.Repository, snapshot SnapshotWithId, ownerIsRoot bool) (*dir, error) {
debug.Log("new dir for snapshot %v (%v)", snapshot.ID.Str(), snapshot.Tree.Str())
tree, err := repo.LoadTree(*snapshot.Tree)
tree, err := repo.LoadTree(ctx, *snapshot.Tree)
if err != nil {
debug.Log(" loadTree(%v) failed: %v", snapshot.ID.Str(), err)
return nil, err
}
items := make(map[string]*restic.Node)
for _, n := range tree.Nodes {
nodes, err := replaceSpecialNodes(repo, n)
nodes, err := replaceSpecialNodes(ctx, repo, n)
if err != nil {
debug.Log(" replaceSpecialNodes(%v) failed: %v", n, err)
return nil, err
@ -167,7 +167,7 @@ func (d *dir) Lookup(ctx context.Context, name string) (fs.Node, error) {
}
switch node.Type {
case "dir":
return newDir(d.repo, node, d.ownerIsRoot)
return newDir(ctx, d.repo, node, d.ownerIsRoot)
case "file":
return newFile(d.repo, node, d.ownerIsRoot)
case "symlink":

10
src/restic/fuse/file.go
View File

@ -9,6 +9,8 @@ import (
"restic"
"restic/debug"
scontext "context"
"bazil.org/fuse"
"bazil.org/fuse/fs"
"golang.org/x/net/context"
@ -25,7 +27,7 @@ var _ = fs.HandleReleaser(&file{})
// for fuse operations.
type BlobLoader interface {
LookupBlobSize(restic.ID, restic.BlobType) (uint, error)
LoadBlob(restic.BlobType, restic.ID, []byte) (int, error)
LoadBlob(scontext.Context, restic.BlobType, restic.ID, []byte) (int, error)
}
type file struct {
@ -88,7 +90,7 @@ func (f *file) Attr(ctx context.Context, a *fuse.Attr) error {
}
func (f *file) getBlobAt(i int) (blob []byte, err error) {
func (f *file) getBlobAt(ctx context.Context, i int) (blob []byte, err error) {
debug.Log("getBlobAt(%v, %v)", f.node.Name, i)
if f.blobs[i] != nil {
return f.blobs[i], nil
@ -100,7 +102,7 @@ func (f *file) getBlobAt(i int) (blob []byte, err error) {
}
buf := restic.NewBlobBuffer(f.sizes[i])
n, err := f.repo.LoadBlob(restic.DataBlob, f.node.Content[i], buf)
n, err := f.repo.LoadBlob(ctx, restic.DataBlob, f.node.Content[i], buf)
if err != nil {
debug.Log("LoadBlob(%v, %v) failed: %v", f.node.Name, f.node.Content[i], err)
return nil, err
@ -137,7 +139,7 @@ func (f *file) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadR
readBytes := 0
remainingBytes := req.Size
for i := startContent; remainingBytes > 0 && i < len(f.sizes); i++ {
blob, err := f.getBlobAt(i)
blob, err := f.getBlobAt(ctx, i)
if err != nil {
return err
}

6
src/restic/fuse/snapshot.go
View File

@ -73,14 +73,14 @@ func (sn *SnapshotsDir) updateCache(ctx context.Context) error {
sn.Lock()
defer sn.Unlock()
for id := range sn.repo.List(restic.SnapshotFile, ctx.Done()) {
for id := range sn.repo.List(ctx, restic.SnapshotFile) {
if sn.processed.Has(id) {
debug.Log("skipping snapshot %v, already in list", id.Str())
continue
}
debug.Log("found snapshot id %v", id.Str())
snapshot, err := restic.LoadSnapshot(sn.repo, id)
snapshot, err := restic.LoadSnapshot(ctx, sn.repo, id)
if err != nil {
return err
}
@ -158,5 +158,5 @@ func (sn *SnapshotsDir) Lookup(ctx context.Context, name string) (fs.Node, error
}
}
return newDirFromSnapshot(sn.repo, snapshot, sn.ownerIsRoot)
return newDirFromSnapshot(ctx, sn.repo, snapshot, sn.ownerIsRoot)
}

29
src/restic/index/index.go
View File

@ -2,6 +2,7 @@
package index
import (
"context"
"fmt"
"os"
"restic"
@ -33,15 +34,12 @@ func newIndex() *Index {
}
// New creates a new index for repo from scratch.
func New(repo restic.Repository, p *restic.Progress) (*Index, error) {
done := make(chan struct{})
defer close(done)
func New(ctx context.Context, repo restic.Repository, p *restic.Progress) (*Index, error) {
p.Start()
defer p.Done()
ch := make(chan worker.Job)
go list.AllPacks(repo, ch, done)
go list.AllPacks(ctx, repo, ch)
idx := newIndex()
@ -84,11 +82,11 @@ type indexJSON struct {
Packs []*packJSON `json:"packs"`
}
func loadIndexJSON(repo restic.Repository, id restic.ID) (*indexJSON, error) {
func loadIndexJSON(ctx context.Context, repo restic.Repository, id restic.ID) (*indexJSON, error) {
debug.Log("process index %v\n", id.Str())
var idx indexJSON
err := repo.LoadJSONUnpacked(restic.IndexFile, id, &idx)
err := repo.LoadJSONUnpacked(ctx, restic.IndexFile, id, &idx)
if err != nil {
return nil, err
}
@ -97,25 +95,22 @@ func loadIndexJSON(repo restic.Repository, id restic.ID) (*indexJSON, error) {
}
// Load creates an index by loading all index files from the repo.
func Load(repo restic.Repository, p *restic.Progress) (*Index, error) {
func Load(ctx context.Context, repo restic.Repository, p *restic.Progress) (*Index, error) {
debug.Log("loading indexes")
p.Start()
defer p.Done()
done := make(chan struct{})
defer close(done)
supersedes := make(map[restic.ID]restic.IDSet)
results := make(map[restic.ID]map[restic.ID]Pack)
index := newIndex()
for id := range repo.List(restic.IndexFile, done) {
for id := range repo.List(ctx, restic.IndexFile) {
p.Report(restic.Stat{Blobs: 1})
debug.Log("Load index %v", id.Str())
idx, err := loadIndexJSON(repo, id)
idx, err := loadIndexJSON(ctx, repo, id)
if err != nil {
return nil, err
}
@ -250,17 +245,17 @@ func (idx *Index) FindBlob(h restic.BlobHandle) (result []Location, err error) {
}
// Save writes the complete index to the repo.
func (idx *Index) Save(repo restic.Repository, supersedes restic.IDs) (restic.ID, error) {
func (idx *Index) Save(ctx context.Context, repo restic.Repository, supersedes restic.IDs) (restic.ID, error) {
packs := make(map[restic.ID][]restic.Blob, len(idx.Packs))
for id, p := range idx.Packs {
packs[id] = p.Entries
}
return Save(repo, packs, supersedes)
return Save(ctx, repo, packs, supersedes)
}
// Save writes a new index containing the given packs.
func Save(repo restic.Repository, packs map[restic.ID][]restic.Blob, supersedes restic.IDs) (restic.ID, error) {
func Save(ctx context.Context, repo restic.Repository, packs map[restic.ID][]restic.Blob, supersedes restic.IDs) (restic.ID, error) {
idx := &indexJSON{
Supersedes: supersedes,
Packs: make([]*packJSON, 0, len(packs)),
@ -285,5 +280,5 @@ func Save(repo restic.Repository, packs map[restic.ID][]restic.Blob, supersedes
idx.Packs = append(idx.Packs, p)
}
return repo.SaveJSONUnpacked(restic.IndexFile, idx)
return repo.SaveJSONUnpacked(ctx, restic.IndexFile, idx)
}

17
src/restic/list/list.go
View File

@ -1,6 +1,7 @@
package list
import (
"context"
"restic"
"restic/worker"
)
@ -9,8 +10,8 @@ const listPackWorkers = 10
// Lister combines lists packs in a repo and blobs in a pack.
type Lister interface {
List(restic.FileType, <-chan struct{}) <-chan restic.ID
ListPack(restic.ID) ([]restic.Blob, int64, error)
List(context.Context, restic.FileType) <-chan restic.ID
ListPack(context.Context, restic.ID) ([]restic.Blob, int64, error)
}
// Result is returned in the channel from LoadBlobsFromAllPacks.
@ -36,10 +37,10 @@ func (l Result) Entries() []restic.Blob {
}
// AllPacks sends the contents of all packs to ch.
func AllPacks(repo Lister, ch chan<- worker.Job, done <-chan struct{}) {
f := func(job worker.Job, done <-chan struct{}) (interface{}, error) {
func AllPacks(ctx context.Context, repo Lister, ch chan<- worker.Job) {
f := func(ctx context.Context, job worker.Job) (interface{}, error) {
packID := job.Data.(restic.ID)
entries, size, err := repo.ListPack(packID)
entries, size, err := repo.ListPack(ctx, packID)
return Result{
packID: packID,
@ -49,14 +50,14 @@ func AllPacks(repo Lister, ch chan<- worker.Job, done <-chan struct{}) {
}
jobCh := make(chan worker.Job)
wp := worker.New(listPackWorkers, f, jobCh, ch)
wp := worker.New(ctx, listPackWorkers, f, jobCh, ch)
go func() {
defer close(jobCh)
for id := range repo.List(restic.DataFile, done) {
for id := range repo.List(ctx, restic.DataFile) {
select {
case jobCh <- worker.Job{Data: id}:
case <-done:
case <-ctx.Done():
return
}
}

49
src/restic/lock.go
View File

@ -59,15 +59,15 @@ func IsAlreadyLocked(err error) bool {
// NewLock returns a new, non-exclusive lock for the repository. If an
// exclusive lock is already held by another process, ErrAlreadyLocked is
// returned.
func NewLock(repo Repository) (*Lock, error) {
return newLock(repo, false)
func NewLock(ctx context.Context, repo Repository) (*Lock, error) {
return newLock(ctx, repo, false)
}
// NewExclusiveLock returns a new, exclusive lock for the repository. If
// another lock (normal and exclusive) is already held by another process,
// ErrAlreadyLocked is returned.
func NewExclusiveLock(repo Repository) (*Lock, error) {
return newLock(repo, true)
func NewExclusiveLock(ctx context.Context, repo Repository) (*Lock, error) {
return newLock(ctx, repo, true)
}
var waitBeforeLockCheck = 200 * time.Millisecond
@ -78,7 +78,7 @@ func TestSetLockTimeout(t testing.TB, d time.Duration) {
waitBeforeLockCheck = d
}
func newLock(repo Repository, excl bool) (*Lock, error) {
func newLock(ctx context.Context, repo Repository, excl bool) (*Lock, error) {
lock := &Lock{
Time: time.Now(),
PID: os.Getpid(),
@ -95,11 +95,11 @@ func newLock(repo Repository, excl bool) (*Lock, error) {
return nil, err
}
if err = lock.checkForOtherLocks(); err != nil {
if err = lock.checkForOtherLocks(ctx); err != nil {
return nil, err
}
lockID, err := lock.createLock()
lockID, err := lock.createLock(ctx)
if err != nil {
return nil, err
}
@ -108,7 +108,7 @@ func newLock(repo Repository, excl bool) (*Lock, error) {
time.Sleep(waitBeforeLockCheck)
if err = lock.checkForOtherLocks(); err != nil {
if err = lock.checkForOtherLocks(ctx); err != nil {
lock.Unlock()
return nil, err
}
@ -133,8 +133,8 @@ func (l *Lock) fillUserInfo() error {
// if there are any other locks, regardless if exclusive or not. If a
// non-exclusive lock is to be created, an error is only returned when an
// exclusive lock is found.
func (l *Lock) checkForOtherLocks() error {
return eachLock(l.repo, func(id ID, lock *Lock, err error) error {
func (l *Lock) checkForOtherLocks(ctx context.Context) error {
return eachLock(ctx, l.repo, func(id ID, lock *Lock, err error) error {
if l.lockID != nil && id.Equal(*l.lockID) {
return nil
}
@ -156,12 +156,9 @@ func (l *Lock) checkForOtherLocks() error {
})
}
func eachLock(repo Repository, f func(ID, *Lock, error) error) error {
done := make(chan struct{})
defer close(done)
for id := range repo.List(LockFile, done) {
lock, err := LoadLock(repo, id)
func eachLock(ctx context.Context, repo Repository, f func(ID, *Lock, error) error) error {
for id := range repo.List(ctx, LockFile) {
lock, err := LoadLock(ctx, repo, id)
err = f(id, lock, err)
if err != nil {
return err
@ -172,8 +169,8 @@ func eachLock(repo Repository, f func(ID, *Lock, error) error) error {
}
// createLock acquires the lock by creating a file in the repository.
func (l *Lock) createLock() (ID, error) {
id, err := l.repo.SaveJSONUnpacked(LockFile, l)
func (l *Lock) createLock(ctx context.Context) (ID, error) {
id, err := l.repo.SaveJSONUnpacked(ctx, LockFile, l)
if err != nil {
return ID{}, err
}
@ -228,9 +225,9 @@ func (l *Lock) Stale() bool {
// Refresh refreshes the lock by creating a new file in the backend with a new
// timestamp. Afterwards the old lock is removed.
func (l *Lock) Refresh() error {
func (l *Lock) Refresh(ctx context.Context) error {
debug.Log("refreshing lock %v", l.lockID.Str())
id, err := l.createLock()
id, err := l.createLock(ctx)
if err != nil {
return err
}
@ -271,9 +268,9 @@ func init() {
}
// LoadLock loads and unserializes a lock from a repository.
func LoadLock(repo Repository, id ID) (*Lock, error) {
func LoadLock(ctx context.Context, repo Repository, id ID) (*Lock, error) {
lock := &Lock{}
if err := repo.LoadJSONUnpacked(LockFile, id, lock); err != nil {
if err := repo.LoadJSONUnpacked(ctx, LockFile, id, lock); err != nil {
return nil, err
}
lock.lockID = &id
@ -282,8 +279,8 @@ func LoadLock(repo Repository, id ID) (*Lock, error) {
}
// RemoveStaleLocks deletes all locks detected as stale from the repository.
func RemoveStaleLocks(repo Repository) error {
return eachLock(repo, func(id ID, lock *Lock, err error) error {
func RemoveStaleLocks(ctx context.Context, repo Repository) error {
return eachLock(ctx, repo, func(id ID, lock *Lock, err error) error {
// ignore locks that cannot be loaded
if err != nil {
return nil
@ -298,8 +295,8 @@ func RemoveStaleLocks(repo Repository) error {
}
// RemoveAllLocks removes all locks forcefully.
func RemoveAllLocks(repo Repository) error {
return eachLock(repo, func(id ID, lock *Lock, err error) error {
func RemoveAllLocks(ctx context.Context, repo Repository) error {
return eachLock(ctx, repo, func(id ID, lock *Lock, err error) error {
return repo.Backend().Remove(context.TODO(), Handle{Type: LockFile, Name: id.String()})
})
}

43
src/restic/mock/backend.go
View File

@ -1,6 +1,7 @@
package mock
import (
"context"
"io"
"restic"
@ -10,13 +11,13 @@ import (
// Backend implements a mock backend.
type Backend struct {
CloseFn func() error
SaveFn func(h restic.Handle, rd io.Reader) error
LoadFn func(h restic.Handle, length int, offset int64) (io.ReadCloser, error)
StatFn func(h restic.Handle) (restic.FileInfo, error)
ListFn func(restic.FileType, <-chan struct{}) <-chan string
RemoveFn func(h restic.Handle) error
TestFn func(h restic.Handle) (bool, error)
DeleteFn func() error
SaveFn func(ctx context.Context, h restic.Handle, rd io.Reader) error
LoadFn func(ctx context.Context, h restic.Handle, length int, offset int64) (io.ReadCloser, error)
StatFn func(ctx context.Context, h restic.Handle) (restic.FileInfo, error)
ListFn func(ctx context.Context, t restic.FileType) <-chan string
RemoveFn func(ctx context.Context, h restic.Handle) error
TestFn func(ctx context.Context, h restic.Handle) (bool, error)
DeleteFn func(ctx context.Context) error
LocationFn func() string
}
@ -39,68 +40,68 @@ func (m *Backend) Location() string {
}
// Save data in the backend.
func (m *Backend) Save(h restic.Handle, rd io.Reader) error {
func (m *Backend) Save(ctx context.Context, h restic.Handle, rd io.Reader) error {
if m.SaveFn == nil {
return errors.New("not implemented")
}
return m.SaveFn(h, rd)
return m.SaveFn(ctx, h, rd)
}
// Load loads data from the backend.
func (m *Backend) Load(h restic.Handle, length int, offset int64) (io.ReadCloser, error) {
func (m *Backend) Load(ctx context.Context, h restic.Handle, length int, offset int64) (io.ReadCloser, error) {
if m.LoadFn == nil {
return nil, errors.New("not implemented")
}
return m.LoadFn(h, length, offset)
return m.LoadFn(ctx, h, length, offset)
}
// Stat an object in the backend.
func (m *Backend) Stat(h restic.Handle) (restic.FileInfo, error) {
func (m *Backend) Stat(ctx context.Context, h restic.Handle) (restic.FileInfo, error) {
if m.StatFn == nil {
return restic.FileInfo{}, errors.New("not implemented")
}
return m.StatFn(h)
return m.StatFn(ctx, h)
}
// List items of type t.
func (m *Backend) List(t restic.FileType, done <-chan struct{}) <-chan string {
func (m *Backend) List(ctx context.Context, t restic.FileType) <-chan string {
if m.ListFn == nil {
ch := make(chan string)
close(ch)
return ch
}
return m.ListFn(t, done)
return m.ListFn(ctx, t)
}
// Remove data from the backend.
func (m *Backend) Remove(h restic.Handle) error {
func (m *Backend) Remove(ctx context.Context, h restic.Handle) error {
if m.RemoveFn == nil {
return errors.New("not implemented")
}
return m.RemoveFn(h)
return m.RemoveFn(ctx, h)
}
// Test for the existence of a specific item.
func (m *Backend) Test(h restic.Handle) (bool, error) {
func (m *Backend) Test(ctx context.Context, h restic.Handle) (bool, error) {
if m.TestFn == nil {
return false, errors.New("not implemented")
}
return m.TestFn(h)
return m.TestFn(ctx, h)
}
// Delete all data.
func (m *Backend) Delete() error {
func (m *Backend) Delete(ctx context.Context) error {
if m.DeleteFn == nil {
return errors.New("not implemented")
}
return m.DeleteFn()
return m.DeleteFn(ctx)
}
// Make sure that Backend implements the backend interface.

9
src/restic/node.go
View File

@ -1,6 +1,7 @@
package restic
import (
"context"
"encoding/json"
"fmt"
"os"
@ -116,7 +117,7 @@ func (node Node) GetExtendedAttribute(a string) []byte {
}
// CreateAt creates the node at the given path and restores all the meta data.
func (node *Node) CreateAt(path string, repo Repository, idx *HardlinkIndex) error {
func (node *Node) CreateAt(ctx context.Context, path string, repo Repository, idx *HardlinkIndex) error {
debug.Log("create node %v at %v", node.Name, path)
switch node.Type {
@ -125,7 +126,7 @@ func (node *Node) CreateAt(path string, repo Repository, idx *HardlinkIndex) err
return err
}
case "file":
if err := node.createFileAt(path, repo, idx); err != nil {
if err := node.createFileAt(ctx, path, repo, idx); err != nil {
return err
}
case "symlink":
@ -228,7 +229,7 @@ func (node Node) createDirAt(path string) error {
return nil
}
func (node Node) createFileAt(path string, repo Repository, idx *HardlinkIndex) error {
func (node Node) createFileAt(ctx context.Context, path string, repo Repository, idx *HardlinkIndex) error {
if node.Links > 1 && idx.Has(node.Inode, node.DeviceID) {
if err := fs.Remove(path); !os.IsNotExist(err) {
return errors.Wrap(err, "RemoveCreateHardlink")
@ -259,7 +260,7 @@ func (node Node) createFileAt(path string, repo Repository, idx *HardlinkIndex)
buf = NewBlobBuffer(int(size))
}
n, err := repo.LoadBlob(DataBlob, id, buf)
n, err := repo.LoadBlob(ctx, DataBlob, id, buf)
if err != nil {
return err
}

5
src/restic/pack/pack_test.go
View File

@ -2,6 +2,7 @@ package pack_test
import (
"bytes"
"context"
"crypto/rand"
"crypto/sha256"
"encoding/binary"
@ -126,7 +127,7 @@ func TestUnpackReadSeeker(t *testing.T) {
id := restic.Hash(packData)
handle := restic.Handle{Type: restic.DataFile, Name: id.String()}
OK(t, b.Save(handle, bytes.NewReader(packData)))
OK(t, b.Save(context.TODO(), handle, bytes.NewReader(packData)))
verifyBlobs(t, bufs, k, restic.ReaderAt(b, handle), packSize)
}
@ -139,6 +140,6 @@ func TestShortPack(t *testing.T) {
id := restic.Hash(packData)
handle := restic.Handle{Type: restic.DataFile, Name: id.String()}
OK(t, b.Save(handle, bytes.NewReader(packData)))
OK(t, b.Save(context.TODO(), handle, bytes.NewReader(packData)))
verifyBlobs(t, bufs, k, restic.ReaderAt(b, handle), packSize)
}

21
src/restic/pipe/pipe.go
View File

@ -1,6 +1,7 @@
package pipe
import (
"context"
"fmt"
"os"
"path/filepath"
@ -78,7 +79,7 @@ func readDirNames(dirname string) ([]string, error) {
// dirs). If false is returned, files are ignored and dirs are not even walked.
type SelectFunc func(item string, fi os.FileInfo) bool
func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs chan<- Job, res chan<- Result) (excluded bool) {
func walk(ctx context.Context, basedir, dir string, selectFunc SelectFunc, jobs chan<- Job, res chan<- Result) (excluded bool) {
debug.Log("start on %q, basedir %q", dir, basedir)
relpath, err := filepath.Rel(basedir, dir)
@ -92,7 +93,7 @@ func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs
debug.Log("error for %v: %v, res %p", dir, err, res)
select {
case jobs <- Dir{basedir: basedir, path: relpath, info: info, error: err, result: res}:
case <-done:
case <-ctx.Done():
}
return
}
@ -107,7 +108,7 @@ func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs
debug.Log("sending file job for %v, res %p", dir, res)
select {
case jobs <- Entry{info: info, basedir: basedir, path: relpath, result: res}:
case <-done:
case <-ctx.Done():
}
return
}
@ -117,7 +118,7 @@ func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs
if err != nil {
debug.Log("Readdirnames(%v) returned error: %v, res %p", dir, err, res)
select {
case <-done:
case <-ctx.Done():
case jobs <- Dir{basedir: basedir, path: relpath, info: info, error: err, result: res}:
}
return
@ -146,7 +147,7 @@ func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs
debug.Log("sending file job for %v, err %v, res %p", subpath, err, res)
select {
case jobs <- Entry{info: fi, error: statErr, basedir: basedir, path: filepath.Join(relpath, name), result: ch}:
case <-done:
case <-ctx.Done():
return
}
continue
@ -156,13 +157,13 @@ func walk(basedir, dir string, selectFunc SelectFunc, done <-chan struct{}, jobs
// between walk and open
debug.RunHook("pipe.walk2", filepath.Join(relpath, name))
walk(basedir, subpath, selectFunc, done, jobs, ch)
walk(ctx, basedir, subpath, selectFunc, jobs, ch)
}
debug.Log("sending dirjob for %q, basedir %q, res %p", dir, basedir, res)
select {
case jobs <- Dir{basedir: basedir, path: relpath, info: info, Entries: entries, result: res}:
case <-done:
case <-ctx.Done():
}
return
@ -191,7 +192,7 @@ func cleanupPath(path string) ([]string, error) {
// Walk sends a Job for each file and directory it finds below the paths. When
// the channel done is closed, processing stops.
func Walk(walkPaths []string, selectFunc SelectFunc, done chan struct{}, jobs chan<- Job, res chan<- Result) {
func Walk(ctx context.Context, walkPaths []string, selectFunc SelectFunc, jobs chan<- Job, res chan<- Result) {
var paths []string
for _, p := range walkPaths {
@ -215,7 +216,7 @@ func Walk(walkPaths []string, selectFunc SelectFunc, done chan struct{}, jobs ch
for _, path := range paths {
debug.Log("start walker for %v", path)
ch := make(chan Result, 1)
excluded := walk(filepath.Dir(path), path, selectFunc, done, jobs, ch)
excluded := walk(ctx, filepath.Dir(path), path, selectFunc, jobs, ch)
if excluded {
debug.Log("walker for %v done, it was excluded by the filter", path)
@ -228,7 +229,7 @@ func Walk(walkPaths []string, selectFunc SelectFunc, done chan struct{}, jobs ch
debug.Log("sending root node, res %p", res)
select {
case <-done:
case <-ctx.Done():
return
case jobs <- Dir{Entries: entries, result: res}:
}

6
src/restic/readerat.go
View File

@ -12,7 +12,7 @@ type backendReaderAt struct {
}
func (brd backendReaderAt) ReadAt(p []byte, offset int64) (n int, err error) {
return ReadAt(brd.be, brd.h, offset, p)
return ReadAt(context.TODO(), brd.be, brd.h, offset, p)
}
// ReaderAt returns an io.ReaderAt for a file in the backend.
@ -21,9 +21,9 @@ func ReaderAt(be Backend, h Handle) io.ReaderAt {
}
// ReadAt reads from the backend handle h at the given position.
func ReadAt(be Backend, h Handle, offset int64, p []byte) (n int, err error) {
func ReadAt(ctx context.Context, be Backend, h Handle, offset int64, p []byte) (n int, err error) {
debug.Log("ReadAt(%v) at %v, len %v", h, offset, len(p))
rd, err := be.Load(context.TODO(), h, len(p), offset)
rd, err := be.Load(ctx, h, len(p), offset)
if err != nil {
return 0, err
}

26
src/restic/repository.go
View File

@ -17,30 +17,30 @@ type Repository interface {
SetIndex(Index)
Index() Index
SaveFullIndex() error
SaveIndex() error
LoadIndex() error
SaveFullIndex(context.Context) error
SaveIndex(context.Context) error
LoadIndex(context.Context) error
Config() Config
LookupBlobSize(ID, BlobType) (uint, error)
List(FileType, <-chan struct{}) <-chan ID
ListPack(ID) ([]Blob, int64, error)
List(context.Context, FileType) <-chan ID
ListPack(context.Context, ID) ([]Blob, int64, error)
Flush() error
SaveUnpacked(FileType, []byte) (ID, error)
SaveJSONUnpacked(FileType, interface{}) (ID, error)
SaveUnpacked(context.Context, FileType, []byte) (ID, error)
SaveJSONUnpacked(context.Context, FileType, interface{}) (ID, error)
LoadJSONUnpacked(FileType, ID, interface{}) error
LoadAndDecrypt(FileType, ID) ([]byte, error)
LoadJSONUnpacked(context.Context, FileType, ID, interface{}) error
LoadAndDecrypt(context.Context, FileType, ID) ([]byte, error)
LoadBlob(BlobType, ID, []byte) (int, error)
SaveBlob(BlobType, []byte, ID) (ID, error)
LoadBlob(context.Context, BlobType, ID, []byte) (int, error)
SaveBlob(context.Context, BlobType, []byte, ID) (ID, error)
LoadTree(ID) (*Tree, error)
SaveTree(t *Tree) (ID, error)
LoadTree(context.Context, ID) (*Tree, error)
SaveTree(context.Context, *Tree) (ID, error)
}
// Deleter removes all data stored in a backend/repo.

5
src/restic/repository/index.go
View File

@ -1,6 +1,7 @@
package repository
import (
"context"
"encoding/json"
"io"
"restic"
@ -519,10 +520,10 @@ func DecodeOldIndex(buf []byte) (idx *Index, err error) {
}
// LoadIndexWithDecoder loads the index and decodes it with fn.
func LoadIndexWithDecoder(repo restic.Repository, id restic.ID, fn func([]byte) (*Index, error)) (idx *Index, err error) {
func LoadIndexWithDecoder(ctx context.Context, repo restic.Repository, id restic.ID, fn func([]byte) (*Index, error)) (idx *Index, err error) {
debug.Log("Loading index %v", id.Str())
buf, err := repo.LoadAndDecrypt(restic.IndexFile, id)
buf, err := repo.LoadAndDecrypt(ctx, restic.IndexFile, id)
if err != nil {
return nil, err
}

23
src/restic/repository/key.go
View File

@ -2,6 +2,7 @@ package repository
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
@ -58,12 +59,12 @@ var (
// createMasterKey creates a new master key in the given backend and encrypts
// it with the password.
func createMasterKey(s *Repository, password string) (*Key, error) {
return AddKey(s, password, nil)
return AddKey(context.TODO(), s, password, nil)
}
// OpenKey tries do decrypt the key specified by name with the given password.
func OpenKey(s *Repository, name string, password string) (*Key, error) {
k, err := LoadKey(s, name)
func OpenKey(ctx context.Context, s *Repository, name string, password string) (*Key, error) {
k, err := LoadKey(ctx, s, name)
if err != nil {
debug.Log("LoadKey(%v) returned error %v", name[:12], err)
return nil, err
@ -113,19 +114,17 @@ func OpenKey(s *Repository, name string, password string) (*Key, error) {
// given password. If none could be found, ErrNoKeyFound is returned. When
// maxKeys is reached, ErrMaxKeysReached is returned. When setting maxKeys to
// zero, all keys in the repo are checked.
func SearchKey(s *Repository, password string, maxKeys int) (*Key, error) {
func SearchKey(ctx context.Context, s *Repository, password string, maxKeys int) (*Key, error) {
checked := 0
// try at most maxKeysForSearch keys in repo
done := make(chan struct{})
defer close(done)
for name := range s.Backend().List(restic.KeyFile, done) {
for name := range s.Backend().List(ctx, restic.KeyFile) {
if maxKeys > 0 && checked > maxKeys {
return nil, ErrMaxKeysReached
}
debug.Log("trying key %v", name[:12])
key, err := OpenKey(s, name, password)
key, err := OpenKey(ctx, s, name, password)
if err != nil {
debug.Log("key %v returned error %v", name[:12], err)
@ -145,9 +144,9 @@ func SearchKey(s *Repository, password string, maxKeys int) (*Key, error) {
}
// LoadKey loads a key from the backend.
func LoadKey(s *Repository, name string) (k *Key, err error) {
func LoadKey(ctx context.Context, s *Repository, name string) (k *Key, err error) {
h := restic.Handle{Type: restic.KeyFile, Name: name}
data, err := backend.LoadAll(s.be, h)
data, err := backend.LoadAll(ctx, s.be, h)
if err != nil {
return nil, err
}
@ -162,7 +161,7 @@ func LoadKey(s *Repository, name string) (k *Key, err error) {
}
// AddKey adds a new key to an already existing repository.
func AddKey(s *Repository, password string, template *crypto.Key) (*Key, error) {
func AddKey(ctx context.Context, s *Repository, password string, template *crypto.Key) (*Key, error) {
// make sure we have valid KDF parameters
if KDFParams == nil {
p, err := crypto.Calibrate(KDFTimeout, KDFMemory)
@ -233,7 +232,7 @@ func AddKey(s *Repository, password string, template *crypto.Key) (*Key, error)
Name: restic.Hash(buf).String(),
}
err = s.be.Save(h, bytes.NewReader(buf))
err = s.be.Save(ctx, h, bytes.NewReader(buf))
if err != nil {
return nil, err
}

5
src/restic/repository/packer_manager.go
View File

@ -1,6 +1,7 @@
package repository
import (
"context"
"crypto/sha256"
"io"
"os"
@ -18,7 +19,7 @@ import (
// Saver implements saving data in a backend.
type Saver interface {
Save(restic.Handle, io.Reader) error
Save(context.Context, restic.Handle, io.Reader) error
}
// Packer holds a pack.Packer together with a hash writer.
@ -118,7 +119,7 @@ func (r *Repository) savePacker(p *Packer) error {
id := restic.IDFromHash(p.hw.Sum(nil))
h := restic.Handle{Type: restic.DataFile, Name: id.String()}
err = r.be.Save(h, p.tmpfile)
err = r.be.Save(context.TODO(), h, p.tmpfile)
if err != nil {
debug.Log("Save(%v) error: %v", h, err)
return err

27
src/restic/repository/parallel.go
View File

@ -1,6 +1,7 @@
package repository
import (
"context"
"restic"
"sync"
@ -18,24 +19,19 @@ func closeIfOpen(ch chan struct{}) {
}
// ParallelWorkFunc gets one file ID to work on. If an error is returned,
// processing stops. If done is closed, the function should return.
type ParallelWorkFunc func(id string, done <-chan struct{}) error
// processing stops. When the contect is cancelled the function should return.
type ParallelWorkFunc func(ctx context.Context, id string) error
// ParallelIDWorkFunc gets one restic.ID to work on. If an error is returned,
// processing stops. If done is closed, the function should return.
type ParallelIDWorkFunc func(id restic.ID, done <-chan struct{}) error
// processing stops. When the context is cancelled the function should return.
type ParallelIDWorkFunc func(ctx context.Context, id restic.ID) error
// FilesInParallel runs n workers of f in parallel, on the IDs that
// repo.List(t) yield. If f returns an error, the process is aborted and the
// first error is returned.
func FilesInParallel(repo restic.Lister, t restic.FileType, n uint, f ParallelWorkFunc) error {
done := make(chan struct{})
defer closeIfOpen(done)
func FilesInParallel(ctx context.Context, repo restic.Lister, t restic.FileType, n uint, f ParallelWorkFunc) error {
wg := &sync.WaitGroup{}
ch := repo.List(t, done)
ch := repo.List(ctx, t)
errors := make(chan error, n)
for i := 0; uint(i) < n; i++ {
@ -50,13 +46,12 @@ func FilesInParallel(repo restic.Lister, t restic.FileType, n uint, f ParallelWo
return
}
err := f(id, done)
err := f(ctx, id)
if err != nil {
closeIfOpen(done)
errors <- err
return
}
case <-done:
case <-ctx.Done():
return
}
}
@ -79,13 +74,13 @@ func FilesInParallel(repo restic.Lister, t restic.FileType, n uint, f ParallelWo
// function that takes a string. Filenames that do not parse as a restic.ID
// are ignored.
func ParallelWorkFuncParseID(f ParallelIDWorkFunc) ParallelWorkFunc {
return func(s string, done <-chan struct{}) error {
return func(ctx context.Context, s string) error {
id, err := restic.ParseID(s)
if err != nil {
debug.Log("invalid ID %q: %v", id, err)
return err
}
return f(id, done)
return f(ctx, id)
}
}

9
src/restic/repository/repack.go
View File

@ -1,6 +1,7 @@
package repository
import (
"context"
"crypto/sha256"
"io"
"restic"
@ -17,7 +18,7 @@ import (
// these packs. Each pack is loaded and the blobs listed in keepBlobs is saved
// into a new pack. Afterwards, the packs are removed. This operation requires
// an exclusive lock on the repo.
func Repack(repo restic.Repository, packs restic.IDSet, keepBlobs restic.BlobSet, p *restic.Progress) (err error) {
func Repack(ctx context.Context, repo restic.Repository, packs restic.IDSet, keepBlobs restic.BlobSet, p *restic.Progress) (err error) {
debug.Log("repacking %d packs while keeping %d blobs", len(packs), len(keepBlobs))
for packID := range packs {
@ -29,7 +30,7 @@ func Repack(repo restic.Repository, packs restic.IDSet, keepBlobs restic.BlobSet
return errors.Wrap(err, "TempFile")
}
beRd, err := repo.Backend().Load(h, 0, 0)
beRd, err := repo.Backend().Load(ctx, h, 0, 0)
if err != nil {
return err
}
@ -100,7 +101,7 @@ func Repack(repo restic.Repository, packs restic.IDSet, keepBlobs restic.BlobSet
h, tempfile.Name(), id)
}
_, err = repo.SaveBlob(entry.Type, buf, entry.ID)
_, err = repo.SaveBlob(ctx, entry.Type, buf, entry.ID)
if err != nil {
return err
}
@ -128,7 +129,7 @@ func Repack(repo restic.Repository, packs restic.IDSet, keepBlobs restic.BlobSet
for packID := range packs {
h := restic.Handle{Type: restic.DataFile, Name: packID.String()}
err := repo.Backend().Remove(h)
err := repo.Backend().Remove(ctx, h)
if err != nil {
debug.Log("error removing pack %v: %v", packID.Str(), err)
return err

101
src/restic/repository/repository.go
View File

@ -2,6 +2,7 @@ package repository
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
@ -50,11 +51,11 @@ func (r *Repository) PrefixLength(t restic.FileType) (int, error) {
// LoadAndDecrypt loads and decrypts data identified by t and id from the
// backend.
func (r *Repository) LoadAndDecrypt(t restic.FileType, id restic.ID) ([]byte, error) {
func (r *Repository) LoadAndDecrypt(ctx context.Context, t restic.FileType, id restic.ID) ([]byte, error) {
debug.Log("load %v with id %v", t, id.Str())
h := restic.Handle{Type: t, Name: id.String()}
buf, err := backend.LoadAll(r.be, h)
buf, err := backend.LoadAll(ctx, r.be, h)
if err != nil {
debug.Log("error loading %v: %v", h, err)
return nil, err
@ -76,7 +77,7 @@ func (r *Repository) LoadAndDecrypt(t restic.FileType, id restic.ID) ([]byte, er
// loadBlob tries to load and decrypt content identified by t and id from a
// pack from the backend, the result is stored in plaintextBuf, which must be
// large enough to hold the complete blob.
func (r *Repository) loadBlob(id restic.ID, t restic.BlobType, plaintextBuf []byte) (int, error) {
func (r *Repository) loadBlob(ctx context.Context, id restic.ID, t restic.BlobType, plaintextBuf []byte) (int, error) {
debug.Log("load %v with id %v (buf len %v, cap %d)", t, id.Str(), len(plaintextBuf), cap(plaintextBuf))
// lookup packs
@ -103,7 +104,7 @@ func (r *Repository) loadBlob(id restic.ID, t restic.BlobType, plaintextBuf []by
plaintextBuf = plaintextBuf[:blob.Length]
n, err := restic.ReadAt(r.be, h, int64(blob.Offset), plaintextBuf)
n, err := restic.ReadAt(ctx, r.be, h, int64(blob.Offset), plaintextBuf)
if err != nil {
debug.Log("error loading blob %v: %v", blob, err)
lastError = err
@ -143,8 +144,8 @@ func (r *Repository) loadBlob(id restic.ID, t restic.BlobType, plaintextBuf []by
// LoadJSONUnpacked decrypts the data and afterwards calls json.Unmarshal on
// the item.
func (r *Repository) LoadJSONUnpacked(t restic.FileType, id restic.ID, item interface{}) (err error) {
buf, err := r.LoadAndDecrypt(t, id)
func (r *Repository) LoadJSONUnpacked(ctx context.Context, t restic.FileType, id restic.ID, item interface{}) (err error) {
buf, err := r.LoadAndDecrypt(ctx, t, id)
if err != nil {
return err
}
@ -159,7 +160,7 @@ func (r *Repository) LookupBlobSize(id restic.ID, tpe restic.BlobType) (uint, er
// SaveAndEncrypt encrypts data and stores it to the backend as type t. If data
// is small enough, it will be packed together with other small blobs.
func (r *Repository) SaveAndEncrypt(t restic.BlobType, data []byte, id *restic.ID) (restic.ID, error) {
func (r *Repository) SaveAndEncrypt(ctx context.Context, t restic.BlobType, data []byte, id *restic.ID) (restic.ID, error) {
if id == nil {
// compute plaintext hash
hashedID := restic.Hash(data)
@ -204,19 +205,19 @@ func (r *Repository) SaveAndEncrypt(t restic.BlobType, data []byte, id *restic.I
// SaveJSONUnpacked serialises item as JSON and encrypts and saves it in the
// backend as type t, without a pack. It returns the storage hash.
func (r *Repository) SaveJSONUnpacked(t restic.FileType, item interface{}) (restic.ID, error) {
func (r *Repository) SaveJSONUnpacked(ctx context.Context, t restic.FileType, item interface{}) (restic.ID, error) {
debug.Log("save new blob %v", t)
plaintext, err := json.Marshal(item)
if err != nil {
return restic.ID{}, errors.Wrap(err, "json.Marshal")
}
return r.SaveUnpacked(t, plaintext)
return r.SaveUnpacked(ctx, t, plaintext)
}
// SaveUnpacked encrypts data and stores it in the backend. Returned is the
// storage hash.
func (r *Repository) SaveUnpacked(t restic.FileType, p []byte) (id restic.ID, err error) {
func (r *Repository) SaveUnpacked(ctx context.Context, t restic.FileType, p []byte) (id restic.ID, err error) {
ciphertext := restic.NewBlobBuffer(len(p))
ciphertext, err = r.Encrypt(ciphertext, p)
if err != nil {
@ -226,7 +227,7 @@ func (r *Repository) SaveUnpacked(t restic.FileType, p []byte) (id restic.ID, er
id = restic.Hash(ciphertext)
h := restic.Handle{Type: t, Name: id.String()}
err = r.be.Save(h, bytes.NewReader(ciphertext))
err = r.be.Save(ctx, h, bytes.NewReader(ciphertext))
if err != nil {
debug.Log("error saving blob %v: %v", h, err)
return restic.ID{}, err
@ -269,7 +270,7 @@ func (r *Repository) SetIndex(i restic.Index) {
}
// SaveIndex saves an index in the repository.
func SaveIndex(repo restic.Repository, index *Index) (restic.ID, error) {
func SaveIndex(ctx context.Context, repo restic.Repository, index *Index) (restic.ID, error) {
buf := bytes.NewBuffer(nil)
err := index.Finalize(buf)
@ -277,15 +278,15 @@ func SaveIndex(repo restic.Repository, index *Index) (restic.ID, error) {
return restic.ID{}, err
}
return repo.SaveUnpacked(restic.IndexFile, buf.Bytes())
return repo.SaveUnpacked(ctx, restic.IndexFile, buf.Bytes())
}
// saveIndex saves all indexes in the backend.
func (r *Repository) saveIndex(indexes ...*Index) error {
func (r *Repository) saveIndex(ctx context.Context, indexes ...*Index) error {
for i, idx := range indexes {
debug.Log("Saving index %d", i)
sid, err := SaveIndex(r, idx)
sid, err := SaveIndex(ctx, r, idx)
if err != nil {
return err
}
@ -297,34 +298,34 @@ func (r *Repository) saveIndex(indexes ...*Index) error {
}
// SaveIndex saves all new indexes in the backend.
func (r *Repository) SaveIndex() error {
return r.saveIndex(r.idx.NotFinalIndexes()...)
func (r *Repository) SaveIndex(ctx context.Context) error {
return r.saveIndex(ctx, r.idx.NotFinalIndexes()...)
}
// SaveFullIndex saves all full indexes in the backend.
func (r *Repository) SaveFullIndex() error {
return r.saveIndex(r.idx.FullIndexes()...)
func (r *Repository) SaveFullIndex(ctx context.Context) error {
return r.saveIndex(ctx, r.idx.FullIndexes()...)
}
const loadIndexParallelism = 20
// LoadIndex loads all index files from the backend in parallel and stores them
// in the master index. The first error that occurred is returned.
func (r *Repository) LoadIndex() error {
func (r *Repository) LoadIndex(ctx context.Context) error {
debug.Log("Loading index")
errCh := make(chan error, 1)
indexes := make(chan *Index)
worker := func(id restic.ID, done <-chan struct{}) error {
idx, err := LoadIndex(r, id)
worker := func(ctx context.Context, id restic.ID) error {
idx, err := LoadIndex(ctx, r, id)
if err != nil {
return err
}
select {
case indexes <- idx:
case <-done:
case <-ctx.Done():
}
return nil
@ -332,7 +333,7 @@ func (r *Repository) LoadIndex() error {
go func() {
defer close(indexes)
errCh <- FilesInParallel(r.be, restic.IndexFile, loadIndexParallelism,
errCh <- FilesInParallel(ctx, r.be, restic.IndexFile, loadIndexParallelism,
ParallelWorkFuncParseID(worker))
}()
@ -348,15 +349,15 @@ func (r *Repository) LoadIndex() error {
}
// LoadIndex loads the index id from backend and returns it.
func LoadIndex(repo restic.Repository, id restic.ID) (*Index, error) {
idx, err := LoadIndexWithDecoder(repo, id, DecodeIndex)
func LoadIndex(ctx context.Context, repo restic.Repository, id restic.ID) (*Index, error) {
idx, err := LoadIndexWithDecoder(ctx, repo, id, DecodeIndex)
if err == nil {
return idx, nil
}
if errors.Cause(err) == ErrOldIndexFormat {
fmt.Fprintf(os.Stderr, "index %v has old format\n", id.Str())
return LoadIndexWithDecoder(repo, id, DecodeOldIndex)
return LoadIndexWithDecoder(ctx, repo, id, DecodeOldIndex)
}
return nil, err
@ -364,8 +365,8 @@ func LoadIndex(repo restic.Repository, id restic.ID) (*Index, error) {
// SearchKey finds a key with the supplied password, afterwards the config is
// read and parsed. It tries at most maxKeys key files in the repo.
func (r *Repository) SearchKey(password string, maxKeys int) error {
key, err := SearchKey(r, password, maxKeys)
func (r *Repository) SearchKey(ctx context.Context, password string, maxKeys int) error {
key, err := SearchKey(ctx, r, password, maxKeys)
if err != nil {
return err
}
@ -373,14 +374,14 @@ func (r *Repository) SearchKey(password string, maxKeys int) error {
r.key = key.master
r.packerManager.key = key.master
r.keyName = key.Name()
r.cfg, err = restic.LoadConfig(r)
r.cfg, err = restic.LoadConfig(ctx, r)
return err
}
// Init creates a new master key with the supplied password, initializes and
// saves the repository config.
func (r *Repository) Init(password string) error {
has, err := r.be.Test(restic.Handle{Type: restic.ConfigFile})
func (r *Repository) Init(ctx context.Context, password string) error {
has, err := r.be.Test(ctx, restic.Handle{Type: restic.ConfigFile})
if err != nil {
return err
}
@ -393,12 +394,12 @@ func (r *Repository) Init(password string) error {
return err
}
return r.init(password, cfg)
return r.init(ctx, password, cfg)
}
// init creates a new master key with the supplied password and uses it to save
// the config into the repo.
func (r *Repository) init(password string, cfg restic.Config) error {
func (r *Repository) init(ctx context.Context, password string, cfg restic.Config) error {
key, err := createMasterKey(r, password)
if err != nil {
return err
@ -408,7 +409,7 @@ func (r *Repository) init(password string, cfg restic.Config) error {
r.packerManager.key = key.master
r.keyName = key.Name()
r.cfg = cfg
_, err = r.SaveJSONUnpacked(restic.ConfigFile, cfg)
_, err = r.SaveJSONUnpacked(ctx, restic.ConfigFile, cfg)
return err
}
@ -443,15 +444,15 @@ func (r *Repository) KeyName() string {
}
// List returns a channel that yields all IDs of type t in the backend.
func (r *Repository) List(t restic.FileType, done <-chan struct{}) <-chan restic.ID {
func (r *Repository) List(ctx context.Context, t restic.FileType) <-chan restic.ID {
out := make(chan restic.ID)
go func() {
defer close(out)
for strID := range r.be.List(t, done) {
for strID := range r.be.List(ctx, t) {
if id, err := restic.ParseID(strID); err == nil {
select {
case out <- id:
case <-done:
case <-ctx.Done():
return
}
}
@ -462,10 +463,10 @@ func (r *Repository) List(t restic.FileType, done <-chan struct{}) <-chan restic
// ListPack returns the list of blobs saved in the pack id and the length of
// the file as stored in the backend.
func (r *Repository) ListPack(id restic.ID) ([]restic.Blob, int64, error) {
func (r *Repository) ListPack(ctx context.Context, id restic.ID) ([]restic.Blob, int64, error) {
h := restic.Handle{Type: restic.DataFile, Name: id.String()}
blobInfo, err := r.Backend().Stat(h)
blobInfo, err := r.Backend().Stat(ctx, h)
if err != nil {
return nil, 0, err
}
@ -480,9 +481,9 @@ func (r *Repository) ListPack(id restic.ID) ([]restic.Blob, int64, error) {
// Delete calls backend.Delete() if implemented, and returns an error
// otherwise.
func (r *Repository) Delete() error {
func (r *Repository) Delete(ctx context.Context) error {
if b, ok := r.be.(restic.Deleter); ok {
return b.Delete()
return b.Delete(ctx)
}
return errors.New("Delete() called for backend that does not implement this method")
@ -496,7 +497,7 @@ func (r *Repository) Close() error {
// LoadBlob loads a blob of type t from the repository to the buffer. buf must
// be large enough to hold the encrypted blob, since it is used as scratch
// space.
func (r *Repository) LoadBlob(t restic.BlobType, id restic.ID, buf []byte) (int, error) {
func (r *Repository) LoadBlob(ctx context.Context, t restic.BlobType, id restic.ID, buf []byte) (int, error) {
debug.Log("load blob %v into buf (len %v, cap %v)", id.Str(), len(buf), cap(buf))
size, err := r.idx.LookupSize(id, t)
if err != nil {
@ -507,7 +508,7 @@ func (r *Repository) LoadBlob(t restic.BlobType, id restic.ID, buf []byte) (int,
return 0, errors.Errorf("buffer is too small for data blob (%d < %d)", cap(buf), restic.CiphertextLength(int(size)))
}
n, err := r.loadBlob(id, t, buf)
n, err := r.loadBlob(ctx, id, t, buf)
if err != nil {
return 0, err
}
@ -520,16 +521,16 @@ func (r *Repository) LoadBlob(t restic.BlobType, id restic.ID, buf []byte) (int,
// SaveBlob saves a blob of type t into the repository. If id is the null id, it
// will be computed and returned.
func (r *Repository) SaveBlob(t restic.BlobType, buf []byte, id restic.ID) (restic.ID, error) {
func (r *Repository) SaveBlob(ctx context.Context, t restic.BlobType, buf []byte, id restic.ID) (restic.ID, error) {
var i *restic.ID
if !id.IsNull() {
i = &id
}
return r.SaveAndEncrypt(t, buf, i)
return r.SaveAndEncrypt(ctx, t, buf, i)
}
// LoadTree loads a tree from the repository.
func (r *Repository) LoadTree(id restic.ID) (*restic.Tree, error) {
func (r *Repository) LoadTree(ctx context.Context, id restic.ID) (*restic.Tree, error) {
debug.Log("load tree %v", id.Str())
size, err := r.idx.LookupSize(id, restic.TreeBlob)
@ -540,7 +541,7 @@ func (r *Repository) LoadTree(id restic.ID) (*restic.Tree, error) {
debug.Log("size is %d, create buffer", size)
buf := restic.NewBlobBuffer(int(size))
n, err := r.loadBlob(id, restic.TreeBlob, buf)
n, err := r.loadBlob(ctx, id, restic.TreeBlob, buf)
if err != nil {
return nil, err
}
@ -558,7 +559,7 @@ func (r *Repository) LoadTree(id restic.ID) (*restic.Tree, error) {
// SaveTree stores a tree into the repository and returns the ID. The ID is
// checked against the index. The tree is only stored when the index does not
// contain the ID.
func (r *Repository) SaveTree(t *restic.Tree) (restic.ID, error) {
func (r *Repository) SaveTree(ctx context.Context, t *restic.Tree) (restic.ID, error) {
buf, err := json.Marshal(t)
if err != nil {
return restic.ID{}, errors.Wrap(err, "MarshalJSON")
@ -573,6 +574,6 @@ func (r *Repository) SaveTree(t *restic.Tree) (restic.ID, error) {
return id, nil
}
_, err = r.SaveBlob(restic.TreeBlob, buf, id)
_, err = r.SaveBlob(ctx, restic.TreeBlob, buf, id)
return id, err
}

5
src/restic/repository/testing.go
View File

@ -1,6 +1,7 @@
package repository
import (
"context"
"os"
"restic"
"restic/backend/local"
@ -50,7 +51,7 @@ func TestRepositoryWithBackend(t testing.TB, be restic.Backend) (r restic.Reposi
repo := New(be)
cfg := restic.TestCreateConfig(t, testChunkerPol)
err := repo.init(test.TestPassword, cfg)
err := repo.init(context.TODO(), test.TestPassword, cfg)
if err != nil {
t.Fatalf("TestRepository(): initialize repo failed: %v", err)
}
@ -94,7 +95,7 @@ func TestOpenLocal(t testing.TB, dir string) (r restic.Repository) {
}
repo := New(be)
err = repo.SearchKey(test.TestPassword, 10)
err = repo.SearchKey(context.TODO(), test.TestPassword, 10)
if err != nil {
t.Fatal(err)
}

21
src/restic/restorer.go
View File

@ -1,6 +1,7 @@
package restic
import (
"context"
"os"
"path/filepath"
@ -30,7 +31,7 @@ func NewRestorer(repo Repository, id ID) (*Restorer, error) {
var err error
r.sn, err = LoadSnapshot(repo, id)
r.sn, err = LoadSnapshot(context.TODO(), repo, id)
if err != nil {
return nil, err
}
@ -38,8 +39,8 @@ func NewRestorer(repo Repository, id ID) (*Restorer, error) {
return r, nil
}
func (res *Restorer) restoreTo(dst string, dir string, treeID ID, idx *HardlinkIndex) error {
tree, err := res.repo.LoadTree(treeID)
func (res *Restorer) restoreTo(ctx context.Context, dst string, dir string, treeID ID, idx *HardlinkIndex) error {
tree, err := res.repo.LoadTree(ctx, treeID)
if err != nil {
return res.Error(dir, nil, err)
}
@ -50,7 +51,7 @@ func (res *Restorer) restoreTo(dst string, dir string, treeID ID, idx *HardlinkI
debug.Log("SelectForRestore returned %v", selectedForRestore)
if selectedForRestore {
err := res.restoreNodeTo(node, dir, dst, idx)
err := res.restoreNodeTo(ctx, node, dir, dst, idx)
if err != nil {
return err
}
@ -62,7 +63,7 @@ func (res *Restorer) restoreTo(dst string, dir string, treeID ID, idx *HardlinkI
}
subp := filepath.Join(dir, node.Name)
err = res.restoreTo(dst, subp, *node.Subtree, idx)
err = res.restoreTo(ctx, dst, subp, *node.Subtree, idx)
if err != nil {
err = res.Error(subp, node, err)
if err != nil {
@ -83,11 +84,11 @@ func (res *Restorer) restoreTo(dst string, dir string, treeID ID, idx *HardlinkI
return nil
}
func (res *Restorer) restoreNodeTo(node *Node, dir string, dst string, idx *HardlinkIndex) error {
func (res *Restorer) restoreNodeTo(ctx context.Context, node *Node, dir string, dst string, idx *HardlinkIndex) error {
debug.Log("node %v, dir %v, dst %v", node.Name, dir, dst)
dstPath := filepath.Join(dst, dir, node.Name)
err := node.CreateAt(dstPath, res.repo, idx)
err := node.CreateAt(ctx, dstPath, res.repo, idx)
if err != nil {
debug.Log("node.CreateAt(%s) error %v", dstPath, err)
}
@ -99,7 +100,7 @@ func (res *Restorer) restoreNodeTo(node *Node, dir string, dst string, idx *Hard
// Create parent directories and retry
err = fs.MkdirAll(filepath.Dir(dstPath), 0700)
if err == nil || os.IsExist(errors.Cause(err)) {
err = node.CreateAt(dstPath, res.repo, idx)
err = node.CreateAt(ctx, dstPath, res.repo, idx)
}
}
@ -118,9 +119,9 @@ func (res *Restorer) restoreNodeTo(node *Node, dir string, dst string, idx *Hard
// RestoreTo creates the directories and files in the snapshot below dst.
// Before an item is created, res.Filter is called.
func (res *Restorer) RestoreTo(dst string) error {
func (res *Restorer) RestoreTo(ctx context.Context, dst string) error {
idx := NewHardlinkIndex()
return res.restoreTo(dst, string(filepath.Separator), *res.sn.Tree, idx)
return res.restoreTo(ctx, dst, string(filepath.Separator), *res.sn.Tree, idx)
}
// Snapshot returns the snapshot this restorer is configured to use.

20
src/restic/snapshot.go
View File

@ -1,6 +1,7 @@
package restic
import (
"context"
"fmt"
"os/user"
"path/filepath"
@ -51,9 +52,9 @@ func NewSnapshot(paths []string, tags []string, hostname string) (*Snapshot, err
}
// LoadSnapshot loads the snapshot with the id and returns it.
func LoadSnapshot(repo Repository, id ID) (*Snapshot, error) {
func LoadSnapshot(ctx context.Context, repo Repository, id ID) (*Snapshot, error) {
sn := &Snapshot{id: &id}
err := repo.LoadJSONUnpacked(SnapshotFile, id, sn)
err := repo.LoadJSONUnpacked(ctx, SnapshotFile, id, sn)
if err != nil {
return nil, err
}
@ -62,12 +63,9 @@ func LoadSnapshot(repo Repository, id ID) (*Snapshot, error) {
}
// LoadAllSnapshots returns a list of all snapshots in the repo.
func LoadAllSnapshots(repo Repository) (snapshots []*Snapshot, err error) {
done := make(chan struct{})
defer close(done)
for id := range repo.List(SnapshotFile, done) {
sn, err := LoadSnapshot(repo, id)
func LoadAllSnapshots(ctx context.Context, repo Repository) (snapshots []*Snapshot, err error) {
for id := range repo.List(ctx, SnapshotFile) {
sn, err := LoadSnapshot(ctx, repo, id)
if err != nil {
return nil, err
}
@ -178,15 +176,15 @@ func (sn *Snapshot) SamePaths(paths []string) bool {
var ErrNoSnapshotFound = errors.New("no snapshot found")
// FindLatestSnapshot finds latest snapshot with optional target/directory, tags and hostname filters.
func FindLatestSnapshot(repo Repository, targets []string, tags []string, hostname string) (ID, error) {
func FindLatestSnapshot(ctx context.Context, repo Repository, targets []string, tags []string, hostname string) (ID, error) {
var (
latest time.Time
latestID ID
found bool
)
for snapshotID := range repo.List(SnapshotFile, make(chan struct{})) {
snapshot, err := LoadSnapshot(repo, snapshotID)
for snapshotID := range repo.List(ctx, SnapshotFile) {
snapshot, err := LoadSnapshot(ctx, repo, snapshotID)
if err != nil {
return ID{}, errors.Errorf("Error listing snapshot: %v", err)
}

19
src/restic/testing.go
View File

@ -1,6 +1,7 @@
package restic
import (
"context"
"encoding/json"
"fmt"
"io"
@ -29,7 +30,7 @@ type fakeFileSystem struct {
// saveFile reads from rd and saves the blobs in the repository. The list of
// IDs is returned.
func (fs *fakeFileSystem) saveFile(rd io.Reader) (blobs IDs) {
func (fs *fakeFileSystem) saveFile(ctx context.Context, rd io.Reader) (blobs IDs) {
if fs.buf == nil {
fs.buf = make([]byte, chunker.MaxSize)
}
@ -53,7 +54,7 @@ func (fs *fakeFileSystem) saveFile(rd io.Reader) (blobs IDs) {
id := Hash(chunk.Data)
if !fs.blobIsKnown(id, DataBlob) {
_, err := fs.repo.SaveBlob(DataBlob, chunk.Data, id)
_, err := fs.repo.SaveBlob(ctx, DataBlob, chunk.Data, id)
if err != nil {
fs.t.Fatalf("error saving chunk: %v", err)
}
@ -103,7 +104,7 @@ func (fs *fakeFileSystem) blobIsKnown(id ID, t BlobType) bool {
}
// saveTree saves a tree of fake files in the repo and returns the ID.
func (fs *fakeFileSystem) saveTree(seed int64, depth int) ID {
func (fs *fakeFileSystem) saveTree(ctx context.Context, seed int64, depth int) ID {
rnd := rand.NewSource(seed)
numNodes := int(rnd.Int63() % maxNodes)
@ -113,7 +114,7 @@ func (fs *fakeFileSystem) saveTree(seed int64, depth int) ID {
// randomly select the type of the node, either tree (p = 1/4) or file (p = 3/4).
if depth > 1 && rnd.Int63()%4 == 0 {
treeSeed := rnd.Int63() % maxSeed
id := fs.saveTree(treeSeed, depth-1)
id := fs.saveTree(ctx, treeSeed, depth-1)
node := &Node{
Name: fmt.Sprintf("dir-%v", treeSeed),
@ -136,7 +137,7 @@ func (fs *fakeFileSystem) saveTree(seed int64, depth int) ID {
Size: uint64(fileSize),
}
node.Content = fs.saveFile(fakeFile(fs.t, fileSeed, fileSize))
node.Content = fs.saveFile(ctx, fakeFile(fs.t, fileSeed, fileSize))
tree.Nodes = append(tree.Nodes, node)
}
@ -145,7 +146,7 @@ func (fs *fakeFileSystem) saveTree(seed int64, depth int) ID {
return id
}
_, err := fs.repo.SaveBlob(TreeBlob, buf, id)
_, err := fs.repo.SaveBlob(ctx, TreeBlob, buf, id)
if err != nil {
fs.t.Fatal(err)
}
@ -176,10 +177,10 @@ func TestCreateSnapshot(t testing.TB, repo Repository, at time.Time, depth int,
duplication: duplication,
}
treeID := fs.saveTree(seed, depth)
treeID := fs.saveTree(context.TODO(), seed, depth)
snapshot.Tree = &treeID
id, err := repo.SaveJSONUnpacked(SnapshotFile, snapshot)
id, err := repo.SaveJSONUnpacked(context.TODO(), SnapshotFile, snapshot)
if err != nil {
t.Fatal(err)
}
@ -193,7 +194,7 @@ func TestCreateSnapshot(t testing.TB, repo Repository, at time.Time, depth int,
t.Fatal(err)
}
err = repo.SaveIndex()
err = repo.SaveIndex(context.TODO())
if err != nil {
t.Fatal(err)
}

31
src/restic/walk/walk.go
View File

@ -1,6 +1,7 @@
package walk
import (
"context"
"fmt"
"os"
"path/filepath"
@ -34,7 +35,7 @@ func NewTreeWalker(ch chan<- loadTreeJob, out chan<- TreeJob) *TreeWalker {
// Walk starts walking the tree given by id. When the channel done is closed,
// processing stops.
func (tw *TreeWalker) Walk(path string, id restic.ID, done chan struct{}) {
func (tw *TreeWalker) Walk(ctx context.Context, path string, id restic.ID) {
debug.Log("starting on tree %v for %v", id.Str(), path)
defer debug.Log("done walking tree %v for %v", id.Str(), path)
@ -48,22 +49,22 @@ func (tw *TreeWalker) Walk(path string, id restic.ID, done chan struct{}) {
if res.err != nil {
select {
case tw.out <- TreeJob{Path: path, Error: res.err}:
case <-done:
case <-ctx.Done():
return
}
return
}
tw.walk(path, res.tree, done)
tw.walk(ctx, path, res.tree)
select {
case tw.out <- TreeJob{Path: path, Tree: res.tree}:
case <-done:
case <-ctx.Done():
return
}
}
func (tw *TreeWalker) walk(path string, tree *restic.Tree, done chan struct{}) {
func (tw *TreeWalker) walk(ctx context.Context, path string, tree *restic.Tree) {
debug.Log("start on %q", path)
defer debug.Log("done for %q", path)
@ -94,7 +95,7 @@ func (tw *TreeWalker) walk(path string, tree *restic.Tree, done chan struct{}) {
res := <-results[i]
if res.err == nil {
tw.walk(p, res.tree, done)
tw.walk(ctx, p, res.tree)
} else {
fmt.Fprintf(os.Stderr, "error loading tree: %v\n", res.err)
}
@ -106,7 +107,7 @@ func (tw *TreeWalker) walk(path string, tree *restic.Tree, done chan struct{}) {
select {
case tw.out <- job:
case <-done:
case <-ctx.Done():
return
}
}
@ -124,14 +125,14 @@ type loadTreeJob struct {
type treeLoader func(restic.ID) (*restic.Tree, error)
func loadTreeWorker(wg *sync.WaitGroup, in <-chan loadTreeJob, load treeLoader, done <-chan struct{}) {
func loadTreeWorker(ctx context.Context, wg *sync.WaitGroup, in <-chan loadTreeJob, load treeLoader) {
debug.Log("start")
defer debug.Log("exit")
defer wg.Done()
for {
select {
case <-done:
case <-ctx.Done():
debug.Log("done channel closed")
return
case job, ok := <-in:
@ -148,7 +149,7 @@ func loadTreeWorker(wg *sync.WaitGroup, in <-chan loadTreeJob, load treeLoader,
select {
case job.res <- loadTreeResult{tree, err}:
debug.Log("job result sent")
case <-done:
case <-ctx.Done():
debug.Log("done channel closed before result could be sent")
return
}
@ -158,7 +159,7 @@ func loadTreeWorker(wg *sync.WaitGroup, in <-chan loadTreeJob, load treeLoader,
// TreeLoader loads tree objects.
type TreeLoader interface {
LoadTree(restic.ID) (*restic.Tree, error)
LoadTree(context.Context, restic.ID) (*restic.Tree, error)
}
const loadTreeWorkers = 10
@ -166,11 +167,11 @@ const loadTreeWorkers = 10
// Tree walks the tree specified by id recursively and sends a job for each
// file and directory it finds. When the channel done is closed, processing
// stops.
func Tree(repo TreeLoader, id restic.ID, done chan struct{}, jobCh chan<- TreeJob) {
func Tree(ctx context.Context, repo TreeLoader, id restic.ID, jobCh chan<- TreeJob) {
debug.Log("start on %v, start workers", id.Str())
load := func(id restic.ID) (*restic.Tree, error) {
tree, err := repo.LoadTree(id)
tree, err := repo.LoadTree(ctx, id)
if err != nil {
return nil, err
}
@ -182,11 +183,11 @@ func Tree(repo TreeLoader, id restic.ID, done chan struct{}, jobCh chan<- TreeJo
var wg sync.WaitGroup
for i := 0; i < loadTreeWorkers; i++ {
wg.Add(1)
go loadTreeWorker(&wg, ch, load, done)
go loadTreeWorker(ctx, &wg, ch, load)
}
tw := NewTreeWalker(ch, jobCh)
tw.Walk("", id, done)
tw.Walk(ctx, "", id)
close(jobCh)
close(ch)

13
src/restic/worker/pool.go
View File

@ -1,5 +1,7 @@
package worker
import "context"
// Job is one unit of work. It is given to a Func, and the returned result and
// error are stored in Result and Error.
type Job struct {
@ -9,12 +11,12 @@ type Job struct {
}
// Func does the actual work within a Pool.
type Func func(job Job, done <-chan struct{}) (result interface{}, err error)
type Func func(ctx context.Context, job Job) (result interface{}, err error)
// Pool implements a worker pool.
type Pool struct {
f Func
done chan struct{}
ctx context.Context
jobCh <-chan Job
resCh chan<- Job
@ -25,10 +27,9 @@ type Pool struct {
// New returns a new worker pool with n goroutines, each running the function
// f. The workers are started immediately.
func New(n int, f Func, jobChan <-chan Job, resultChan chan<- Job) *Pool {
func New(ctx context.Context, n int, f Func, jobChan <-chan Job, resultChan chan<- Job) *Pool {
p := &Pool{
f: f,
done: make(chan struct{}),
workersExit: make(chan struct{}),
allWorkersDone: make(chan struct{}),
numWorkers: n,
@ -75,7 +76,7 @@ func (p *Pool) runWorker(numWorker int) {
for {
select {
case <-p.done:
case <-p.ctx.Done():
return
case job, ok = <-inCh:
@ -83,7 +84,7 @@ func (p *Pool) runWorker(numWorker int) {
return
}
job.Result, job.Error = p.f(job, p.done)
job.Result, job.Error = p.f(p.ctx, job)
inCh = nil
outCh = p.resCh

11
src/restic/worker/pool_test.go
View File

@ -1,6 +1,7 @@
package worker_test
import (
"context"
"testing"
"restic/errors"
@ -12,7 +13,7 @@ const concurrency = 10
var errTooLarge = errors.New("too large")
func square(job worker.Job, done <-chan struct{}) (interface{}, error) {
func square(ctx context.Context, job worker.Job) (interface{}, error) {
n := job.Data.(int)
if n > 2000 {
return nil, errTooLarge
@ -20,15 +21,15 @@ func square(job worker.Job, done <-chan struct{}) (interface{}, error) {
return n * n, nil
}
func newBufferedPool(bufsize int, n int, f worker.Func) (chan worker.Job, chan worker.Job, *worker.Pool) {
func newBufferedPool(ctx context.Context, bufsize int, n int, f worker.Func) (chan worker.Job, chan worker.Job, *worker.Pool) {
inCh := make(chan worker.Job, bufsize)
outCh := make(chan worker.Job, bufsize)
return inCh, outCh, worker.New(n, f, inCh, outCh)
return inCh, outCh, worker.New(ctx, n, f, inCh, outCh)
}
func TestPool(t *testing.T) {
inCh, outCh, p := newBufferedPool(200, concurrency, square)
inCh, outCh, p := newBufferedPool(context.TODO(), 200, concurrency, square)
for i := 0; i < 150; i++ {
inCh <- worker.Job{Data: i}
@ -53,7 +54,7 @@ func TestPool(t *testing.T) {
}
func TestPoolErrors(t *testing.T) {
inCh, outCh, p := newBufferedPool(200, concurrency, square)
inCh, outCh, p := newBufferedPool(context.TODO(), 200, concurrency, square)
for i := 0; i < 150; i++ {
inCh <- worker.Job{Data: i + 1900}