TrueCloudLab/restic
16
0
Fork 2
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

archiver: hide implementation details

Browse source
This commit is contained in:
Michael Eischer 2024-08-27 11:26:52 +02:00
parent e5cdae9c84
commit c6fae0320e
13 changed files with 223 additions and 224 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

106
internal/archiver/archiver.go
View file

@ -75,6 +75,14 @@ type archiverRepo interface {
} }
// Archiver saves a directory structure to the repo. // Archiver saves a directory structure to the repo.
//
// An Archiver has a number of worker goroutines handling saving the different
// data structures to the repository, the details are implemented by the
// fileSaver, blobSaver, and treeSaver types.
//
// The main goroutine (the one calling Snapshot()) traverses the directory tree
// and delegates all work to these worker pools. They return a futureNode which
// can be resolved later, by calling Wait() on it.
type Archiver struct { type Archiver struct {
Repo archiverRepo Repo archiverRepo
SelectByName SelectByNameFunc SelectByName SelectByNameFunc
@ -82,9 +90,9 @@ type Archiver struct {
FS fs.FS FS fs.FS
Options Options Options Options
blobSaver *BlobSaver blobSaver *blobSaver
fileSaver *FileSaver fileSaver *fileSaver
treeSaver *TreeSaver treeSaver *treeSaver
mu sync.Mutex mu sync.Mutex
summary *Summary summary *Summary
@ -160,7 +168,7 @@ func (o Options) ApplyDefaults() Options {
if o.SaveTreeConcurrency == 0 { if o.SaveTreeConcurrency == 0 {
// can either wait for a file, wait for a tree, serialize a tree or wait for saveblob // can either wait for a file, wait for a tree, serialize a tree or wait for saveblob
// the last two are cpu-bound and thus mutually exclusive. // the last two are cpu-bound and thus mutually exclusive.
// Also allow waiting for FileReadConcurrency files, this is the maximum of FutureFiles // Also allow waiting for FileReadConcurrency files, this is the maximum of files
// which currently can be in progress. The main backup loop blocks when trying to queue // which currently can be in progress. The main backup loop blocks when trying to queue
// more files to read. // more files to read.
o.SaveTreeConcurrency = uint(runtime.GOMAXPROCS(0)) + o.ReadConcurrency o.SaveTreeConcurrency = uint(runtime.GOMAXPROCS(0)) + o.ReadConcurrency
@ -297,27 +305,27 @@ func (arch *Archiver) wrapLoadTreeError(id restic.ID, err error) error {
// saveDir stores a directory in the repo and returns the node. snPath is the // saveDir stores a directory in the repo and returns the node. snPath is the
// path within the current snapshot. // path within the current snapshot.
func (arch *Archiver) saveDir(ctx context.Context, snPath string, dir string, fi os.FileInfo, previous *restic.Tree, complete CompleteFunc) (d FutureNode, err error) { func (arch *Archiver) saveDir(ctx context.Context, snPath string, dir string, fi os.FileInfo, previous *restic.Tree, complete fileCompleteFunc) (d futureNode, err error) {
debug.Log("%v %v", snPath, dir) debug.Log("%v %v", snPath, dir)
treeNode, err := arch.nodeFromFileInfo(snPath, dir, fi, false) treeNode, err := arch.nodeFromFileInfo(snPath, dir, fi, false)
if err != nil { if err != nil {
return FutureNode{}, err return futureNode{}, err
} }
names, err := fs.Readdirnames(arch.FS, dir, fs.O_NOFOLLOW) names, err := fs.Readdirnames(arch.FS, dir, fs.O_NOFOLLOW)
if err != nil { if err != nil {
return FutureNode{}, err return futureNode{}, err
} }
sort.Strings(names) sort.Strings(names)
nodes := make([]FutureNode, 0, len(names)) nodes := make([]futureNode, 0, len(names))
for _, name := range names { for _, name := range names {
// test if context has been cancelled // test if context has been cancelled
if ctx.Err() != nil { if ctx.Err() != nil {
debug.Log("context has been cancelled, aborting") debug.Log("context has been cancelled, aborting")
return FutureNode{}, ctx.Err() return futureNode{}, ctx.Err()
} }
pathname := arch.FS.Join(dir, name) pathname := arch.FS.Join(dir, name)
@ -333,7 +341,7 @@ func (arch *Archiver) saveDir(ctx context.Context, snPath string, dir string, fi
continue continue
} }
return FutureNode{}, err return futureNode{}, err
} }
if excluded { if excluded {
@ -348,11 +356,11 @@ func (arch *Archiver) saveDir(ctx context.Context, snPath string, dir string, fi
return fn, nil return fn, nil
} }
// FutureNode holds a reference to a channel that returns a FutureNodeResult // futureNode holds a reference to a channel that returns a FutureNodeResult
// or a reference to an already existing result. If the result is available // or a reference to an already existing result. If the result is available
// immediately, then storing a reference directly requires less memory than // immediately, then storing a reference directly requires less memory than
// using the indirection via a channel. // using the indirection via a channel.
type FutureNode struct { type futureNode struct {
ch <-chan futureNodeResult ch <-chan futureNodeResult
res *futureNodeResult res *futureNodeResult
} }
@ -365,18 +373,18 @@ type futureNodeResult struct {
err error err error
} }
func newFutureNode() (FutureNode, chan<- futureNodeResult) { func newFutureNode() (futureNode, chan<- futureNodeResult) {
ch := make(chan futureNodeResult, 1) ch := make(chan futureNodeResult, 1)
return FutureNode{ch: ch}, ch return futureNode{ch: ch}, ch
} }
func newFutureNodeWithResult(res futureNodeResult) FutureNode { func newFutureNodeWithResult(res futureNodeResult) futureNode {
return FutureNode{ return futureNode{
res: &res, res: &res,
} }
} }
func (fn *FutureNode) take(ctx context.Context) futureNodeResult { func (fn *futureNode) take(ctx context.Context) futureNodeResult {
if fn.res != nil { if fn.res != nil {
res := fn.res res := fn.res
// free result // free result
@ -415,19 +423,19 @@ func (arch *Archiver) allBlobsPresent(previous *restic.Node) bool {
// Errors and completion needs to be handled by the caller. // Errors and completion needs to be handled by the caller.
// //
// snPath is the path within the current snapshot. // snPath is the path within the current snapshot.
func (arch *Archiver) save(ctx context.Context, snPath, target string, previous *restic.Node) (fn FutureNode, excluded bool, err error) { func (arch *Archiver) save(ctx context.Context, snPath, target string, previous *restic.Node) (fn futureNode, excluded bool, err error) {
start := time.Now() start := time.Now()
debug.Log("%v target %q, previous %v", snPath, target, previous) debug.Log("%v target %q, previous %v", snPath, target, previous)
abstarget, err := arch.FS.Abs(target) abstarget, err := arch.FS.Abs(target)
if err != nil { if err != nil {
return FutureNode{}, false, err return futureNode{}, false, err
} }
// exclude files by path before running Lstat to reduce number of lstat calls // exclude files by path before running Lstat to reduce number of lstat calls
if !arch.SelectByName(abstarget) { if !arch.SelectByName(abstarget) {
debug.Log("%v is excluded by path", target) debug.Log("%v is excluded by path", target)
return FutureNode{}, true, nil return futureNode{}, true, nil
} }
// get file info and run remaining select functions that require file information // get file info and run remaining select functions that require file information
@ -436,13 +444,13 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
debug.Log("lstat() for %v returned error: %v", target, err) debug.Log("lstat() for %v returned error: %v", target, err)
err = arch.error(abstarget, err) err = arch.error(abstarget, err)
if err != nil { if err != nil {
return FutureNode{}, false, errors.WithStack(err) return futureNode{}, false, errors.WithStack(err)
} }
return FutureNode{}, true, nil return futureNode{}, true, nil
} }
if !arch.Select(abstarget, fi) { if !arch.Select(abstarget, fi) {
debug.Log("%v is excluded", target) debug.Log("%v is excluded", target)
return FutureNode{}, true, nil return futureNode{}, true, nil
} }
switch { switch {
@ -458,7 +466,7 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
arch.CompleteBlob(previous.Size) arch.CompleteBlob(previous.Size)
node, err := arch.nodeFromFileInfo(snPath, target, fi, false) node, err := arch.nodeFromFileInfo(snPath, target, fi, false)
if err != nil { if err != nil {
return FutureNode{}, false, err return futureNode{}, false, err
} }
// copy list of blobs // copy list of blobs
@ -477,7 +485,7 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
err := errors.Errorf("parts of %v not found in the repository index; storing the file again", target) err := errors.Errorf("parts of %v not found in the repository index; storing the file again", target)
err = arch.error(abstarget, err) err = arch.error(abstarget, err)
if err != nil { if err != nil {
return FutureNode{}, false, err return futureNode{}, false, err
} }
} }
@ -488,9 +496,9 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
debug.Log("Openfile() for %v returned error: %v", target, err) debug.Log("Openfile() for %v returned error: %v", target, err)
err = arch.error(abstarget, err) err = arch.error(abstarget, err)
if err != nil { if err != nil {
return FutureNode{}, false, errors.WithStack(err) return futureNode{}, false, errors.WithStack(err)
} }
return FutureNode{}, true, nil return futureNode{}, true, nil
} }
fi, err = file.Stat() fi, err = file.Stat()
@ -499,9 +507,9 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
_ = file.Close() _ = file.Close()
err = arch.error(abstarget, err) err = arch.error(abstarget, err)
if err != nil { if err != nil {
return FutureNode{}, false, errors.WithStack(err) return futureNode{}, false, errors.WithStack(err)
} }
return FutureNode{}, true, nil return futureNode{}, true, nil
} }
// make sure it's still a file // make sure it's still a file
@ -510,9 +518,9 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
_ = file.Close() _ = file.Close()
err = arch.error(abstarget, err) err = arch.error(abstarget, err)
if err != nil { if err != nil {
return FutureNode{}, false, err return futureNode{}, false, err
} }
return FutureNode{}, true, nil return futureNode{}, true, nil
} }
// Save will close the file, we don't need to do that // Save will close the file, we don't need to do that
@ -533,7 +541,7 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
err = arch.error(abstarget, err) err = arch.error(abstarget, err)
} }
if err != nil { if err != nil {
return FutureNode{}, false, err return futureNode{}, false, err
} }
fn, err = arch.saveDir(ctx, snPath, target, fi, oldSubtree, fn, err = arch.saveDir(ctx, snPath, target, fi, oldSubtree,
@ -542,19 +550,19 @@ func (arch *Archiver) save(ctx context.Context, snPath, target string, previous
}) })
if err != nil { if err != nil {
debug.Log("SaveDir for %v returned error: %v", snPath, err) debug.Log("SaveDir for %v returned error: %v", snPath, err)
return FutureNode{}, false, err return futureNode{}, false, err
} }
case fi.Mode()&os.ModeSocket > 0: case fi.Mode()&os.ModeSocket > 0:
debug.Log(" %v is a socket, ignoring", target) debug.Log(" %v is a socket, ignoring", target)
return FutureNode{}, true, nil return futureNode{}, true, nil
default: default:
debug.Log(" %v other", target) debug.Log(" %v other", target)
node, err := arch.nodeFromFileInfo(snPath, target, fi, false) node, err := arch.nodeFromFileInfo(snPath, target, fi, false)
if err != nil { if err != nil {
return FutureNode{}, false, err return futureNode{}, false, err
} }
fn = newFutureNodeWithResult(futureNodeResult{ fn = newFutureNodeWithResult(futureNodeResult{
snPath: snPath, snPath: snPath,
@ -621,17 +629,17 @@ func (arch *Archiver) statDir(dir string) (os.FileInfo, error) {
// saveTree stores a Tree in the repo, returned is the tree. snPath is the path // saveTree stores a Tree in the repo, returned is the tree. snPath is the path
// within the current snapshot. // within the current snapshot.
func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree, previous *restic.Tree, complete CompleteFunc) (FutureNode, int, error) { func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *tree, previous *restic.Tree, complete fileCompleteFunc) (futureNode, int, error) {
var node *restic.Node var node *restic.Node
if snPath != "/" { if snPath != "/" {
if atree.FileInfoPath == "" { if atree.FileInfoPath == "" {
return FutureNode{}, 0, errors.Errorf("FileInfoPath for %v is empty", snPath) return futureNode{}, 0, errors.Errorf("FileInfoPath for %v is empty", snPath)
} }
fi, err := arch.statDir(atree.FileInfoPath) fi, err := arch.statDir(atree.FileInfoPath)
if err != nil { if err != nil {
return FutureNode{}, 0, err return futureNode{}, 0, err
} }
debug.Log("%v, dir node data loaded from %v", snPath, atree.FileInfoPath) debug.Log("%v, dir node data loaded from %v", snPath, atree.FileInfoPath)
@ -639,7 +647,7 @@ func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree,
// thus ignore errors for such folders. // thus ignore errors for such folders.
node, err = arch.nodeFromFileInfo(snPath, atree.FileInfoPath, fi, true) node, err = arch.nodeFromFileInfo(snPath, atree.FileInfoPath, fi, true)
if err != nil { if err != nil {
return FutureNode{}, 0, err return futureNode{}, 0, err
} }
} else { } else {
// fake root node // fake root node
@ -648,7 +656,7 @@ func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree,
debug.Log("%v (%v nodes), parent %v", snPath, len(atree.Nodes), previous) debug.Log("%v (%v nodes), parent %v", snPath, len(atree.Nodes), previous)
nodeNames := atree.NodeNames() nodeNames := atree.NodeNames()
nodes := make([]FutureNode, 0, len(nodeNames)) nodes := make([]futureNode, 0, len(nodeNames))
// iterate over the nodes of atree in lexicographic (=deterministic) order // iterate over the nodes of atree in lexicographic (=deterministic) order
for _, name := range nodeNames { for _, name := range nodeNames {
@ -656,7 +664,7 @@ func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree,
// test if context has been cancelled // test if context has been cancelled
if ctx.Err() != nil { if ctx.Err() != nil {
return FutureNode{}, 0, ctx.Err() return futureNode{}, 0, ctx.Err()
} }
// this is a leaf node // this is a leaf node
@ -669,11 +677,11 @@ func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree,
// ignore error // ignore error
continue continue
} }
return FutureNode{}, 0, err return futureNode{}, 0, err
} }
if err != nil { if err != nil {
return FutureNode{}, 0, err return futureNode{}, 0, err
} }
if !excluded { if !excluded {
@ -691,7 +699,7 @@ func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree,
err = arch.error(join(snPath, name), err) err = arch.error(join(snPath, name), err)
} }
if err != nil { if err != nil {
return FutureNode{}, 0, err return futureNode{}, 0, err
} }
// not a leaf node, archive subtree // not a leaf node, archive subtree
@ -699,7 +707,7 @@ func (arch *Archiver) saveTree(ctx context.Context, snPath string, atree *Tree,
arch.trackItem(snItem, oldNode, n, is, time.Since(start)) arch.trackItem(snItem, oldNode, n, is, time.Since(start))
}) })
if err != nil { if err != nil {
return FutureNode{}, 0, err return futureNode{}, 0, err
} }
nodes = append(nodes, fn) nodes = append(nodes, fn)
} }
@ -779,16 +787,16 @@ func (arch *Archiver) loadParentTree(ctx context.Context, sn *restic.Snapshot) *
// runWorkers starts the worker pools, which are stopped when the context is cancelled. // runWorkers starts the worker pools, which are stopped when the context is cancelled.
func (arch *Archiver) runWorkers(ctx context.Context, wg *errgroup.Group) { func (arch *Archiver) runWorkers(ctx context.Context, wg *errgroup.Group) {
arch.blobSaver = NewBlobSaver(ctx, wg, arch.Repo, arch.Options.SaveBlobConcurrency) arch.blobSaver = newBlobSaver(ctx, wg, arch.Repo, arch.Options.SaveBlobConcurrency)
arch.fileSaver = NewFileSaver(ctx, wg, arch.fileSaver = newFileSaver(ctx, wg,
arch.blobSaver.Save, arch.blobSaver.Save,
arch.Repo.Config().ChunkerPolynomial, arch.Repo.Config().ChunkerPolynomial,
arch.Options.ReadConcurrency, arch.Options.SaveBlobConcurrency) arch.Options.ReadConcurrency, arch.Options.SaveBlobConcurrency)
arch.fileSaver.CompleteBlob = arch.CompleteBlob arch.fileSaver.CompleteBlob = arch.CompleteBlob
arch.fileSaver.NodeFromFileInfo = arch.nodeFromFileInfo arch.fileSaver.NodeFromFileInfo = arch.nodeFromFileInfo
arch.treeSaver = NewTreeSaver(ctx, wg, arch.Options.SaveTreeConcurrency, arch.blobSaver.Save, arch.Error) arch.treeSaver = newTreeSaver(ctx, wg, arch.Options.SaveTreeConcurrency, arch.blobSaver.Save, arch.Error)
} }
func (arch *Archiver) stopWorkers() { func (arch *Archiver) stopWorkers() {
@ -809,7 +817,7 @@ func (arch *Archiver) Snapshot(ctx context.Context, targets []string, opts Snaps
return nil, restic.ID{}, nil, err return nil, restic.ID{}, nil, err
} }
atree, err := NewTree(arch.FS, cleanTargets) atree, err := newTree(arch.FS, cleanTargets)
if err != nil { if err != nil {
return nil, restic.ID{}, nil, err return nil, restic.ID{}, nil, err
} }

2
internal/archiver/archiver_test.go
View file

@ -1121,7 +1121,7 @@ func TestArchiverSaveTree(t *testing.T) {
test.prepare(t) test.prepare(t)
} }
atree, err := NewTree(testFS, test.targets) atree, err := newTree(testFS, test.targets)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }

34
internal/archiver/blob_saver.go
View file

@ -9,22 +9,22 @@ import (
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
) )
// Saver allows saving a blob. // saver allows saving a blob.
type Saver interface { type saver interface {
SaveBlob(ctx context.Context, t restic.BlobType, data []byte, id restic.ID, storeDuplicate bool) (restic.ID, bool, int, error) SaveBlob(ctx context.Context, t restic.BlobType, data []byte, id restic.ID, storeDuplicate bool) (restic.ID, bool, int, error)
} }
// BlobSaver concurrently saves incoming blobs to the repo. // blobSaver concurrently saves incoming blobs to the repo.
type BlobSaver struct { type blobSaver struct {
repo Saver repo saver
ch chan<- saveBlobJob ch chan<- saveBlobJob
} }
// NewBlobSaver returns a new blob. A worker pool is started, it is stopped // newBlobSaver returns a new blob. A worker pool is started, it is stopped
// when ctx is cancelled. // when ctx is cancelled.
func NewBlobSaver(ctx context.Context, wg *errgroup.Group, repo Saver, workers uint) *BlobSaver { func newBlobSaver(ctx context.Context, wg *errgroup.Group, repo saver, workers uint) *blobSaver {
ch := make(chan saveBlobJob) ch := make(chan saveBlobJob)
s := &BlobSaver{ s := &blobSaver{
repo: repo, repo: repo,
ch: ch, ch: ch,
} }
@ -38,13 +38,13 @@ func NewBlobSaver(ctx context.Context, wg *errgroup.Group, repo Saver, workers u
return s return s
} }
func (s *BlobSaver) TriggerShutdown() { func (s *blobSaver) TriggerShutdown() {
close(s.ch) close(s.ch)
} }
// Save stores a blob in the repo. It checks the index and the known blobs // Save stores a blob in the repo. It checks the index and the known blobs
// before saving anything. It takes ownership of the buffer passed in. // before saving anything. It takes ownership of the buffer passed in.
func (s *BlobSaver) Save(ctx context.Context, t restic.BlobType, buf *Buffer, filename string, cb func(res SaveBlobResponse)) { func (s *blobSaver) Save(ctx context.Context, t restic.BlobType, buf *buffer, filename string, cb func(res saveBlobResponse)) {
select { select {
case s.ch <- saveBlobJob{BlobType: t, buf: buf, fn: filename, cb: cb}: case s.ch <- saveBlobJob{BlobType: t, buf: buf, fn: filename, cb: cb}:
case <-ctx.Done(): case <-ctx.Done():
@ -54,26 +54,26 @@ func (s *BlobSaver) Save(ctx context.Context, t restic.BlobType, buf *Buffer, fi
type saveBlobJob struct { type saveBlobJob struct {
restic.BlobType restic.BlobType
buf *Buffer buf *buffer
fn string fn string
cb func(res SaveBlobResponse) cb func(res saveBlobResponse)
} }
type SaveBlobResponse struct { type saveBlobResponse struct {
id restic.ID id restic.ID
length int length int
sizeInRepo int sizeInRepo int
known bool known bool
} }
func (s *BlobSaver) saveBlob(ctx context.Context, t restic.BlobType, buf []byte) (SaveBlobResponse, error) { func (s *blobSaver) saveBlob(ctx context.Context, t restic.BlobType, buf []byte) (saveBlobResponse, error) {
id, known, sizeInRepo, err := s.repo.SaveBlob(ctx, t, buf, restic.ID{}, false) id, known, sizeInRepo, err := s.repo.SaveBlob(ctx, t, buf, restic.ID{}, false)
if err != nil { if err != nil {
return SaveBlobResponse{}, err return saveBlobResponse{}, err
} }
return SaveBlobResponse{ return saveBlobResponse{
id: id, id: id,
length: len(buf), length: len(buf),
sizeInRepo: sizeInRepo, sizeInRepo: sizeInRepo,
@ -81,7 +81,7 @@ func (s *BlobSaver) saveBlob(ctx context.Context, t restic.BlobType, buf []byte)
}, nil }, nil
} }
func (s *BlobSaver) worker(ctx context.Context, jobs <-chan saveBlobJob) error { func (s *blobSaver) worker(ctx context.Context, jobs <-chan saveBlobJob) error {
for { for {
var job saveBlobJob var job saveBlobJob
var ok bool var ok bool

16
internal/archiver/blob_saver_test.go
View file

@ -38,20 +38,20 @@ func TestBlobSaver(t *testing.T) {
wg, ctx := errgroup.WithContext(ctx) wg, ctx := errgroup.WithContext(ctx)
saver := &saveFail{} saver := &saveFail{}
b := NewBlobSaver(ctx, wg, saver, uint(runtime.NumCPU())) b := newBlobSaver(ctx, wg, saver, uint(runtime.NumCPU()))
var wait sync.WaitGroup var wait sync.WaitGroup
var results []SaveBlobResponse var results []saveBlobResponse
var lock sync.Mutex var lock sync.Mutex
wait.Add(20) wait.Add(20)
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
buf := &Buffer{Data: []byte(fmt.Sprintf("foo%d", i))} buf := &buffer{Data: []byte(fmt.Sprintf("foo%d", i))}
idx := i idx := i
lock.Lock() lock.Lock()
results = append(results, SaveBlobResponse{}) results = append(results, saveBlobResponse{})
lock.Unlock() lock.Unlock()
b.Save(ctx, restic.DataBlob, buf, "file", func(res SaveBlobResponse) { b.Save(ctx, restic.DataBlob, buf, "file", func(res saveBlobResponse) {
lock.Lock() lock.Lock()
results[idx] = res results[idx] = res
lock.Unlock() lock.Unlock()
@ -95,11 +95,11 @@ func TestBlobSaverError(t *testing.T) {
failAt: int32(test.failAt), failAt: int32(test.failAt),
} }
b := NewBlobSaver(ctx, wg, saver, uint(runtime.NumCPU())) b := newBlobSaver(ctx, wg, saver, uint(runtime.NumCPU()))
for i := 0; i < test.blobs; i++ { for i := 0; i < test.blobs; i++ {
buf := &Buffer{Data: []byte(fmt.Sprintf("foo%d", i))} buf := &buffer{Data: []byte(fmt.Sprintf("foo%d", i))}
b.Save(ctx, restic.DataBlob, buf, "errfile", func(res SaveBlobResponse) {}) b.Save(ctx, restic.DataBlob, buf, "errfile", func(res saveBlobResponse) {})
} }
b.TriggerShutdown() b.TriggerShutdown()

26
internal/archiver/buffer.go
View file

@ -1,14 +1,14 @@
package archiver package archiver
// Buffer is a reusable buffer. After the buffer has been used, Release should // buffer is a reusable buffer. After the buffer has been used, Release should
// be called so the underlying slice is put back into the pool. // be called so the underlying slice is put back into the pool.
type Buffer struct { type buffer struct {
Data []byte Data []byte
pool *BufferPool pool *bufferPool
} }
// Release puts the buffer back into the pool it came from. // Release puts the buffer back into the pool it came from.
func (b *Buffer) Release() { func (b *buffer) Release() {
pool := b.pool pool := b.pool
if pool == nil || cap(b.Data) > pool.defaultSize { if pool == nil || cap(b.Data) > pool.defaultSize {
return return
@ -20,32 +20,32 @@ func (b *Buffer) Release() {
} }
} }
// BufferPool implements a limited set of reusable buffers. // bufferPool implements a limited set of reusable buffers.
type BufferPool struct { type bufferPool struct {
ch chan *Buffer ch chan *buffer
defaultSize int defaultSize int
} }
// NewBufferPool initializes a new buffer pool. The pool stores at most max // newBufferPool initializes a new buffer pool. The pool stores at most max
// items. New buffers are created with defaultSize. Buffers that have grown // items. New buffers are created with defaultSize. Buffers that have grown
// larger are not put back. // larger are not put back.
func NewBufferPool(max int, defaultSize int) *BufferPool { func newBufferPool(max int, defaultSize int) *bufferPool {
b := &BufferPool{ b := &bufferPool{
ch: make(chan *Buffer, max), ch: make(chan *buffer, max),
defaultSize: defaultSize, defaultSize: defaultSize,
} }
return b return b
} }
// Get returns a new buffer, either from the pool or newly allocated. // Get returns a new buffer, either from the pool or newly allocated.
func (pool *BufferPool) Get() *Buffer { func (pool *bufferPool) Get() *buffer {
select { select {
case buf := <-pool.ch: case buf := <-pool.ch:
return buf return buf
default: default:
} }
b := &Buffer{ b := &buffer{
Data: make([]byte, pool.defaultSize), Data: make([]byte, pool.defaultSize),
pool: pool, pool: pool,
} }

9
internal/archiver/doc.go
View file

@ -1,12 +1,3 @@
// Package archiver contains the code which reads files, splits them into // Package archiver contains the code which reads files, splits them into
// chunks and saves the data to the repository. // chunks and saves the data to the repository.
//
// An Archiver has a number of worker goroutines handling saving the different
// data structures to the repository, the details are implemented by the
// FileSaver, BlobSaver, and TreeSaver types.
//
// The main goroutine (the one calling Snapshot()) traverses the directory tree
// and delegates all work to these worker pools. They return a type
// (FutureFile, FutureBlob, and FutureTree) which can be resolved later, by
// calling Wait() on it.
package archiver package archiver

36
internal/archiver/file_saver.go
View file

@ -15,13 +15,13 @@ import (
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
) )
// SaveBlobFn saves a blob to a repo. // saveBlobFn saves a blob to a repo.
type SaveBlobFn func(context.Context, restic.BlobType, *Buffer, string, func(res SaveBlobResponse)) type saveBlobFn func(context.Context, restic.BlobType, *buffer, string, func(res saveBlobResponse))
// FileSaver concurrently saves incoming files to the repo. // fileSaver concurrently saves incoming files to the repo.
type FileSaver struct { type fileSaver struct {
saveFilePool *BufferPool saveFilePool *bufferPool
saveBlob SaveBlobFn saveBlob saveBlobFn
pol chunker.Pol pol chunker.Pol
@ -32,18 +32,18 @@ type FileSaver struct {
NodeFromFileInfo func(snPath, filename string, fi os.FileInfo, ignoreXattrListError bool) (*restic.Node, error) NodeFromFileInfo func(snPath, filename string, fi os.FileInfo, ignoreXattrListError bool) (*restic.Node, error)
} }
// NewFileSaver returns a new file saver. A worker pool with fileWorkers is // newFileSaver returns a new file saver. A worker pool with fileWorkers is
// started, it is stopped when ctx is cancelled. // started, it is stopped when ctx is cancelled.
func NewFileSaver(ctx context.Context, wg *errgroup.Group, save SaveBlobFn, pol chunker.Pol, fileWorkers, blobWorkers uint) *FileSaver { func newFileSaver(ctx context.Context, wg *errgroup.Group, save saveBlobFn, pol chunker.Pol, fileWorkers, blobWorkers uint) *fileSaver {
ch := make(chan saveFileJob) ch := make(chan saveFileJob)
debug.Log("new file saver with %v file workers and %v blob workers", fileWorkers, blobWorkers) debug.Log("new file saver with %v file workers and %v blob workers", fileWorkers, blobWorkers)
poolSize := fileWorkers + blobWorkers poolSize := fileWorkers + blobWorkers
s := &FileSaver{ s := &fileSaver{
saveBlob: save, saveBlob: save,
saveFilePool: NewBufferPool(int(poolSize), chunker.MaxSize), saveFilePool: newBufferPool(int(poolSize), chunker.MaxSize),
pol: pol, pol: pol,
ch: ch, ch: ch,
@ -60,18 +60,18 @@ func NewFileSaver(ctx context.Context, wg *errgroup.Group, save SaveBlobFn, pol
return s return s
} }
func (s *FileSaver) TriggerShutdown() { func (s *fileSaver) TriggerShutdown() {
close(s.ch) close(s.ch)
} }
// CompleteFunc is called when the file has been saved. // fileCompleteFunc is called when the file has been saved.
type CompleteFunc func(*restic.Node, ItemStats) type fileCompleteFunc func(*restic.Node, ItemStats)
// Save stores the file f and returns the data once it has been completed. The // Save stores the file f and returns the data once it has been completed. The
// file is closed by Save. completeReading is only called if the file was read // file is closed by Save. completeReading is only called if the file was read
// successfully. complete is always called. If completeReading is called, then // successfully. complete is always called. If completeReading is called, then
// this will always happen before calling complete. // this will always happen before calling complete.
func (s *FileSaver) Save(ctx context.Context, snPath string, target string, file fs.File, fi os.FileInfo, start func(), completeReading func(), complete CompleteFunc) FutureNode { func (s *fileSaver) Save(ctx context.Context, snPath string, target string, file fs.File, fi os.FileInfo, start func(), completeReading func(), complete fileCompleteFunc) futureNode {
fn, ch := newFutureNode() fn, ch := newFutureNode()
job := saveFileJob{ job := saveFileJob{
snPath: snPath, snPath: snPath,
@ -105,11 +105,11 @@ type saveFileJob struct {
start func() start func()
completeReading func() completeReading func()
complete CompleteFunc complete fileCompleteFunc
} }
// saveFile stores the file f in the repo, then closes it. // saveFile stores the file f in the repo, then closes it.
func (s *FileSaver) saveFile(ctx context.Context, chnker *chunker.Chunker, snPath string, target string, f fs.File, fi os.FileInfo, start func(), finishReading func(), finish func(res futureNodeResult)) { func (s *fileSaver) saveFile(ctx context.Context, chnker *chunker.Chunker, snPath string, target string, f fs.File, fi os.FileInfo, start func(), finishReading func(), finish func(res futureNodeResult)) {
start() start()
fnr := futureNodeResult{ fnr := futureNodeResult{
@ -205,7 +205,7 @@ func (s *FileSaver) saveFile(ctx context.Context, chnker *chunker.Chunker, snPat
node.Content = append(node.Content, restic.ID{}) node.Content = append(node.Content, restic.ID{})
lock.Unlock() lock.Unlock()
s.saveBlob(ctx, restic.DataBlob, buf, target, func(sbr SaveBlobResponse) { s.saveBlob(ctx, restic.DataBlob, buf, target, func(sbr saveBlobResponse) {
lock.Lock() lock.Lock()
if !sbr.known { if !sbr.known {
fnr.stats.DataBlobs++ fnr.stats.DataBlobs++
@ -246,7 +246,7 @@ func (s *FileSaver) saveFile(ctx context.Context, chnker *chunker.Chunker, snPat
completeBlob() completeBlob()
} }
func (s *FileSaver) worker(ctx context.Context, jobs <-chan saveFileJob) { func (s *fileSaver) worker(ctx context.Context, jobs <-chan saveFileJob) {
// a worker has one chunker which is reused for each file (because it contains a rather large buffer) // a worker has one chunker which is reused for each file (because it contains a rather large buffer)
chnker := chunker.New(nil, s.pol) chnker := chunker.New(nil, s.pol)

10
internal/archiver/file_saver_test.go
View file

@ -30,11 +30,11 @@ func createTestFiles(t testing.TB, num int) (files []string) {
return files return files
} }
func startFileSaver(ctx context.Context, t testing.TB) (*FileSaver, context.Context, *errgroup.Group) { func startFileSaver(ctx context.Context, t testing.TB) (*fileSaver, context.Context, *errgroup.Group) {
wg, ctx := errgroup.WithContext(ctx) wg, ctx := errgroup.WithContext(ctx)
saveBlob := func(ctx context.Context, tpe restic.BlobType, buf *Buffer, _ string, cb func(SaveBlobResponse)) { saveBlob := func(ctx context.Context, tpe restic.BlobType, buf *buffer, _ string, cb func(saveBlobResponse)) {
cb(SaveBlobResponse{ cb(saveBlobResponse{
id: restic.Hash(buf.Data), id: restic.Hash(buf.Data),
length: len(buf.Data), length: len(buf.Data),
sizeInRepo: len(buf.Data), sizeInRepo: len(buf.Data),
@ -48,7 +48,7 @@ func startFileSaver(ctx context.Context, t testing.TB) (*FileSaver, context.Cont
t.Fatal(err) t.Fatal(err)
} }
s := NewFileSaver(ctx, wg, saveBlob, pol, workers, workers) s := newFileSaver(ctx, wg, saveBlob, pol, workers, workers)
s.NodeFromFileInfo = func(snPath, filename string, fi os.FileInfo, ignoreXattrListError bool) (*restic.Node, error) { s.NodeFromFileInfo = func(snPath, filename string, fi os.FileInfo, ignoreXattrListError bool) (*restic.Node, error) {
return fs.NodeFromFileInfo(filename, fi, ignoreXattrListError) return fs.NodeFromFileInfo(filename, fi, ignoreXattrListError)
} }
@ -69,7 +69,7 @@ func TestFileSaver(t *testing.T) {
testFs := fs.Local{} testFs := fs.Local{}
s, ctx, wg := startFileSaver(ctx, t) s, ctx, wg := startFileSaver(ctx, t)
var results []FutureNode var results []futureNode
for _, filename := range files { for _, filename := range files {
f, err := testFs.OpenFile(filename, os.O_RDONLY, 0) f, err := testFs.OpenFile(filename, os.O_RDONLY, 0)

4
internal/archiver/scanner.go
View file

@ -38,7 +38,7 @@ type ScanStats struct {
Bytes uint64 Bytes uint64
} }
func (s *Scanner) scanTree(ctx context.Context, stats ScanStats, tree Tree) (ScanStats, error) { func (s *Scanner) scanTree(ctx context.Context, stats ScanStats, tree tree) (ScanStats, error) {
// traverse the path in the file system for all leaf nodes // traverse the path in the file system for all leaf nodes
if tree.Leaf() { if tree.Leaf() {
abstarget, err := s.FS.Abs(tree.Path) abstarget, err := s.FS.Abs(tree.Path)
@ -83,7 +83,7 @@ func (s *Scanner) Scan(ctx context.Context, targets []string) error {
debug.Log("clean targets %v", cleanTargets) debug.Log("clean targets %v", cleanTargets)
// we're using the same tree representation as the archiver does // we're using the same tree representation as the archiver does
tree, err := NewTree(s.FS, cleanTargets) tree, err := newTree(s.FS, cleanTargets)
if err != nil { if err != nil {
return err return err
} }

54
internal/archiver/tree.go
View file

@ -9,7 +9,7 @@ import (
"github.com/restic/restic/internal/fs" "github.com/restic/restic/internal/fs"
) )
// Tree recursively defines how a snapshot should look like when // tree recursively defines how a snapshot should look like when
// archived. // archived.
// //
// When `Path` is set, this is a leaf node and the contents of `Path` should be // When `Path` is set, this is a leaf node and the contents of `Path` should be
@ -20,8 +20,8 @@ import (
// //
// `FileInfoPath` is used to extract metadata for intermediate (=non-leaf) // `FileInfoPath` is used to extract metadata for intermediate (=non-leaf)
// trees. // trees.
type Tree struct { type tree struct {
Nodes map[string]Tree Nodes map[string]tree
Path string // where the files/dirs to be saved are found Path string // where the files/dirs to be saved are found
FileInfoPath string // where the dir can be found that is not included itself, but its subdirs FileInfoPath string // where the dir can be found that is not included itself, but its subdirs
Root string // parent directory of the tree Root string // parent directory of the tree
@ -95,13 +95,13 @@ func rootDirectory(fs fs.FS, target string) string {
} }
// Add adds a new file or directory to the tree. // Add adds a new file or directory to the tree.
func (t *Tree) Add(fs fs.FS, path string) error { func (t *tree) Add(fs fs.FS, path string) error {
if path == "" { if path == "" {
panic("invalid path (empty string)") panic("invalid path (empty string)")
} }
if t.Nodes == nil { if t.Nodes == nil {
t.Nodes = make(map[string]Tree) t.Nodes = make(map[string]tree)
} }
pc, virtualPrefix := pathComponents(fs, path, false) pc, virtualPrefix := pathComponents(fs, path, false)
@ -111,7 +111,7 @@ func (t *Tree) Add(fs fs.FS, path string) error {
name := pc[0] name := pc[0]
root := rootDirectory(fs, path) root := rootDirectory(fs, path)
tree := Tree{Root: root} tree := tree{Root: root}
origName := name origName := name
i := 0 i := 0
@ -152,63 +152,63 @@ func (t *Tree) Add(fs fs.FS, path string) error {
} }
// add adds a new target path into the tree. // add adds a new target path into the tree.
func (t *Tree) add(fs fs.FS, target, root string, pc []string) error { func (t *tree) add(fs fs.FS, target, root string, pc []string) error {
if len(pc) == 0 { if len(pc) == 0 {
return errors.Errorf("invalid path %q", target) return errors.Errorf("invalid path %q", target)
} }
if t.Nodes == nil { if t.Nodes == nil {
t.Nodes = make(map[string]Tree) t.Nodes = make(map[string]tree)
} }
name := pc[0] name := pc[0]
if len(pc) == 1 { if len(pc) == 1 {
tree, ok := t.Nodes[name] node, ok := t.Nodes[name]
if !ok { if !ok {
t.Nodes[name] = Tree{Path: target} t.Nodes[name] = tree{Path: target}
return nil return nil
} }
if tree.Path != "" { if node.Path != "" {
return errors.Errorf("path is already set for target %v", target) return errors.Errorf("path is already set for target %v", target)
} }
tree.Path = target node.Path = target
t.Nodes[name] = tree t.Nodes[name] = node
return nil return nil
} }
tree := Tree{} node := tree{}
if other, ok := t.Nodes[name]; ok { if other, ok := t.Nodes[name]; ok {
tree = other node = other
} }
subroot := fs.Join(root, name) subroot := fs.Join(root, name)
tree.FileInfoPath = subroot node.FileInfoPath = subroot
err := tree.add(fs, target, subroot, pc[1:]) err := node.add(fs, target, subroot, pc[1:])
if err != nil { if err != nil {
return err return err
} }
t.Nodes[name] = tree t.Nodes[name] = node
return nil return nil
} }
func (t Tree) String() string { func (t tree) String() string {
return formatTree(t, "") return formatTree(t, "")
} }
// Leaf returns true if this is a leaf node, which means Path is set to a // Leaf returns true if this is a leaf node, which means Path is set to a
// non-empty string and the contents of Path should be inserted at this point // non-empty string and the contents of Path should be inserted at this point
// in the tree. // in the tree.
func (t Tree) Leaf() bool { func (t tree) Leaf() bool {
return t.Path != "" return t.Path != ""
} }
// NodeNames returns the sorted list of subtree names. // NodeNames returns the sorted list of subtree names.
func (t Tree) NodeNames() []string { func (t tree) NodeNames() []string {
// iterate over the nodes of atree in lexicographic (=deterministic) order // iterate over the nodes of atree in lexicographic (=deterministic) order
names := make([]string, 0, len(t.Nodes)) names := make([]string, 0, len(t.Nodes))
for name := range t.Nodes { for name := range t.Nodes {
@ -219,7 +219,7 @@ func (t Tree) NodeNames() []string {
} }
// formatTree returns a text representation of the tree t. // formatTree returns a text representation of the tree t.
func formatTree(t Tree, indent string) (s string) { func formatTree(t tree, indent string) (s string) {
for name, node := range t.Nodes { for name, node := range t.Nodes {
s += fmt.Sprintf("%v/%v, root %q, path %q, meta %q\n", indent, name, node.Root, node.Path, node.FileInfoPath) s += fmt.Sprintf("%v/%v, root %q, path %q, meta %q\n", indent, name, node.Root, node.Path, node.FileInfoPath)
s += formatTree(node, indent+" ") s += formatTree(node, indent+" ")
@ -228,7 +228,7 @@ func formatTree(t Tree, indent string) (s string) {
} }
// unrollTree unrolls the tree so that only leaf nodes have Path set. // unrollTree unrolls the tree so that only leaf nodes have Path set.
func unrollTree(f fs.FS, t *Tree) error { func unrollTree(f fs.FS, t *tree) error {
// if the current tree is a leaf node (Path is set) and has additional // if the current tree is a leaf node (Path is set) and has additional
// nodes, add the contents of Path to the nodes. // nodes, add the contents of Path to the nodes.
if t.Path != "" && len(t.Nodes) > 0 { if t.Path != "" && len(t.Nodes) > 0 {
@ -252,7 +252,7 @@ func unrollTree(f fs.FS, t *Tree) error {
return errors.Errorf("tree unrollTree: collision on path, node %#v, path %q", node, f.Join(t.Path, entry)) return errors.Errorf("tree unrollTree: collision on path, node %#v, path %q", node, f.Join(t.Path, entry))
} }
t.Nodes[entry] = Tree{Path: f.Join(t.Path, entry)} t.Nodes[entry] = tree{Path: f.Join(t.Path, entry)}
} }
t.Path = "" t.Path = ""
} }
@ -269,10 +269,10 @@ func unrollTree(f fs.FS, t *Tree) error {
return nil return nil
} }
// NewTree creates a Tree from the target files/directories. // newTree creates a Tree from the target files/directories.
func NewTree(fs fs.FS, targets []string) (*Tree, error) { func newTree(fs fs.FS, targets []string) (*tree, error) {
debug.Log("targets: %v", targets) debug.Log("targets: %v", targets)
tree := &Tree{} tree := &tree{}
seen := make(map[string]struct{}) seen := make(map[string]struct{})
for _, target := range targets { for _, target := range targets {
target = fs.Clean(target) target = fs.Clean(target)

32
internal/archiver/tree_saver.go
View file

@ -9,20 +9,20 @@ import (
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
) )
// TreeSaver concurrently saves incoming trees to the repo. // treeSaver concurrently saves incoming trees to the repo.
type TreeSaver struct { type treeSaver struct {
saveBlob SaveBlobFn saveBlob saveBlobFn
errFn ErrorFunc errFn ErrorFunc
ch chan<- saveTreeJob ch chan<- saveTreeJob
} }
// NewTreeSaver returns a new tree saver. A worker pool with treeWorkers is // newTreeSaver returns a new tree saver. A worker pool with treeWorkers is
// started, it is stopped when ctx is cancelled. // started, it is stopped when ctx is cancelled.
func NewTreeSaver(ctx context.Context, wg *errgroup.Group, treeWorkers uint, saveBlob SaveBlobFn, errFn ErrorFunc) *TreeSaver { func newTreeSaver(ctx context.Context, wg *errgroup.Group, treeWorkers uint, saveBlob saveBlobFn, errFn ErrorFunc) *treeSaver {
ch := make(chan saveTreeJob) ch := make(chan saveTreeJob)
s := &TreeSaver{ s := &treeSaver{
ch: ch, ch: ch,
saveBlob: saveBlob, saveBlob: saveBlob,
errFn: errFn, errFn: errFn,
@ -37,12 +37,12 @@ func NewTreeSaver(ctx context.Context, wg *errgroup.Group, treeWorkers uint, sav
return s return s
} }
func (s *TreeSaver) TriggerShutdown() { func (s *treeSaver) TriggerShutdown() {
close(s.ch) close(s.ch)
} }
// Save stores the dir d and returns the data once it has been completed. // Save stores the dir d and returns the data once it has been completed.
func (s *TreeSaver) Save(ctx context.Context, snPath string, target string, node *restic.Node, nodes []FutureNode, complete CompleteFunc) FutureNode { func (s *treeSaver) Save(ctx context.Context, snPath string, target string, node *restic.Node, nodes []futureNode, complete fileCompleteFunc) futureNode {
fn, ch := newFutureNode() fn, ch := newFutureNode()
job := saveTreeJob{ job := saveTreeJob{
snPath: snPath, snPath: snPath,
@ -66,13 +66,13 @@ type saveTreeJob struct {
snPath string snPath string
target string target string
node *restic.Node node *restic.Node
nodes []FutureNode nodes []futureNode
ch chan<- futureNodeResult ch chan<- futureNodeResult
complete CompleteFunc complete fileCompleteFunc
} }
// save stores the nodes as a tree in the repo. // save stores the nodes as a tree in the repo.
func (s *TreeSaver) save(ctx context.Context, job *saveTreeJob) (*restic.Node, ItemStats, error) { func (s *treeSaver) save(ctx context.Context, job *saveTreeJob) (*restic.Node, ItemStats, error) {
var stats ItemStats var stats ItemStats
node := job.node node := job.node
nodes := job.nodes nodes := job.nodes
@ -84,7 +84,7 @@ func (s *TreeSaver) save(ctx context.Context, job *saveTreeJob) (*restic.Node, I
for i, fn := range nodes { for i, fn := range nodes {
// fn is a copy, so clear the original value explicitly // fn is a copy, so clear the original value explicitly
nodes[i] = FutureNode{} nodes[i] = futureNode{}
fnr := fn.take(ctx) fnr := fn.take(ctx)
// return the error if it wasn't ignored // return the error if it wasn't ignored
@ -128,9 +128,9 @@ func (s *TreeSaver) save(ctx context.Context, job *saveTreeJob) (*restic.Node, I
return nil, stats, err return nil, stats, err
} }
b := &Buffer{Data: buf} b := &buffer{Data: buf}
ch := make(chan SaveBlobResponse, 1) ch := make(chan saveBlobResponse, 1)
s.saveBlob(ctx, restic.TreeBlob, b, job.target, func(res SaveBlobResponse) { s.saveBlob(ctx, restic.TreeBlob, b, job.target, func(res saveBlobResponse) {
ch <- res ch <- res
}) })
@ -149,7 +149,7 @@ func (s *TreeSaver) save(ctx context.Context, job *saveTreeJob) (*restic.Node, I
} }
} }
func (s *TreeSaver) worker(ctx context.Context, jobs <-chan saveTreeJob) error { func (s *treeSaver) worker(ctx context.Context, jobs <-chan saveTreeJob) error {
for { for {
var job saveTreeJob var job saveTreeJob
var ok bool var ok bool

16
internal/archiver/tree_saver_test.go
View file

@ -12,8 +12,8 @@ import (
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
) )
func treeSaveHelper(_ context.Context, _ restic.BlobType, buf *Buffer, _ string, cb func(res SaveBlobResponse)) { func treeSaveHelper(_ context.Context, _ restic.BlobType, buf *buffer, _ string, cb func(res saveBlobResponse)) {
cb(SaveBlobResponse{ cb(saveBlobResponse{
id: restic.NewRandomID(), id: restic.NewRandomID(),
known: false, known: false,
length: len(buf.Data), length: len(buf.Data),
@ -21,7 +21,7 @@ func treeSaveHelper(_ context.Context, _ restic.BlobType, buf *Buffer, _ string,
}) })
} }
func setupTreeSaver() (context.Context, context.CancelFunc, *TreeSaver, func() error) { func setupTreeSaver() (context.Context, context.CancelFunc, *treeSaver, func() error) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
wg, ctx := errgroup.WithContext(ctx) wg, ctx := errgroup.WithContext(ctx)
@ -29,7 +29,7 @@ func setupTreeSaver() (context.Context, context.CancelFunc, *TreeSaver, func() e
return err return err
} }
b := NewTreeSaver(ctx, wg, uint(runtime.NumCPU()), treeSaveHelper, errFn) b := newTreeSaver(ctx, wg, uint(runtime.NumCPU()), treeSaveHelper, errFn)
shutdown := func() error { shutdown := func() error {
b.TriggerShutdown() b.TriggerShutdown()
@ -43,7 +43,7 @@ func TestTreeSaver(t *testing.T) {
ctx, cancel, b, shutdown := setupTreeSaver() ctx, cancel, b, shutdown := setupTreeSaver()
defer cancel() defer cancel()
var results []FutureNode var results []futureNode
for i := 0; i < 20; i++ { for i := 0; i < 20; i++ {
node := &restic.Node{ node := &restic.Node{
@ -83,13 +83,13 @@ func TestTreeSaverError(t *testing.T) {
ctx, cancel, b, shutdown := setupTreeSaver() ctx, cancel, b, shutdown := setupTreeSaver()
defer cancel() defer cancel()
var results []FutureNode var results []futureNode
for i := 0; i < test.trees; i++ { for i := 0; i < test.trees; i++ {
node := &restic.Node{ node := &restic.Node{
Name: fmt.Sprintf("file-%d", i), Name: fmt.Sprintf("file-%d", i),
} }
nodes := []FutureNode{ nodes := []futureNode{
newFutureNodeWithResult(futureNodeResult{node: &restic.Node{ newFutureNodeWithResult(futureNodeResult{node: &restic.Node{
Name: fmt.Sprintf("child-%d", i), Name: fmt.Sprintf("child-%d", i),
}}), }}),
@ -128,7 +128,7 @@ func TestTreeSaverDuplicates(t *testing.T) {
node := &restic.Node{ node := &restic.Node{
Name: "file", Name: "file",
} }
nodes := []FutureNode{ nodes := []futureNode{
newFutureNodeWithResult(futureNodeResult{node: &restic.Node{ newFutureNodeWithResult(futureNodeResult{node: &restic.Node{
Name: "child", Name: "child",
}}), }}),

102
internal/archiver/tree_test.go
View file

@ -12,7 +12,7 @@ import (
) )
// debug.Log requires Tree.String. // debug.Log requires Tree.String.
var _ fmt.Stringer = Tree{} var _ fmt.Stringer = tree{}
func TestPathComponents(t *testing.T) { func TestPathComponents(t *testing.T) {
var tests = []struct { var tests = []struct {
@ -142,20 +142,20 @@ func TestTree(t *testing.T) {
var tests = []struct { var tests = []struct {
targets []string targets []string
src TestDir src TestDir
want Tree want tree
unix bool unix bool
win bool win bool
mustError bool mustError bool
}{ }{
{ {
targets: []string{"foo"}, targets: []string{"foo"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Path: "foo", Root: "."}, "foo": {Path: "foo", Root: "."},
}}, }},
}, },
{ {
targets: []string{"foo", "bar", "baz"}, targets: []string{"foo", "bar", "baz"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Path: "foo", Root: "."}, "foo": {Path: "foo", Root: "."},
"bar": {Path: "bar", Root: "."}, "bar": {Path: "bar", Root: "."},
"baz": {Path: "baz", Root: "."}, "baz": {Path: "baz", Root: "."},
@ -163,8 +163,8 @@ func TestTree(t *testing.T) {
}, },
{ {
targets: []string{"foo/user1", "foo/user2", "foo/other"}, targets: []string{"foo/user1", "foo/user2", "foo/other"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/user1")}, "user1": {Path: filepath.FromSlash("foo/user1")},
"user2": {Path: filepath.FromSlash("foo/user2")}, "user2": {Path: filepath.FromSlash("foo/user2")},
"other": {Path: filepath.FromSlash("foo/other")}, "other": {Path: filepath.FromSlash("foo/other")},
@ -173,9 +173,9 @@ func TestTree(t *testing.T) {
}, },
{ {
targets: []string{"foo/work/user1", "foo/work/user2"}, targets: []string{"foo/work/user1", "foo/work/user2"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]Tree{ "work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/work/user1")}, "user1": {Path: filepath.FromSlash("foo/work/user1")},
"user2": {Path: filepath.FromSlash("foo/work/user2")}, "user2": {Path: filepath.FromSlash("foo/work/user2")},
}}, }},
@ -184,50 +184,50 @@ func TestTree(t *testing.T) {
}, },
{ {
targets: []string{"foo/user1", "bar/user1", "foo/other"}, targets: []string{"foo/user1", "bar/user1", "foo/other"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/user1")}, "user1": {Path: filepath.FromSlash("foo/user1")},
"other": {Path: filepath.FromSlash("foo/other")}, "other": {Path: filepath.FromSlash("foo/other")},
}}, }},
"bar": {Root: ".", FileInfoPath: "bar", Nodes: map[string]Tree{ "bar": {Root: ".", FileInfoPath: "bar", Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("bar/user1")}, "user1": {Path: filepath.FromSlash("bar/user1")},
}}, }},
}}, }},
}, },
{ {
targets: []string{"../work"}, targets: []string{"../work"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"work": {Root: "..", Path: filepath.FromSlash("../work")}, "work": {Root: "..", Path: filepath.FromSlash("../work")},
}}, }},
}, },
{ {
targets: []string{"../work/other"}, targets: []string{"../work/other"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"work": {Root: "..", FileInfoPath: filepath.FromSlash("../work"), Nodes: map[string]Tree{ "work": {Root: "..", FileInfoPath: filepath.FromSlash("../work"), Nodes: map[string]tree{
"other": {Path: filepath.FromSlash("../work/other")}, "other": {Path: filepath.FromSlash("../work/other")},
}}, }},
}}, }},
}, },
{ {
targets: []string{"foo/user1", "../work/other", "foo/user2"}, targets: []string{"foo/user1", "../work/other", "foo/user2"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/user1")}, "user1": {Path: filepath.FromSlash("foo/user1")},
"user2": {Path: filepath.FromSlash("foo/user2")}, "user2": {Path: filepath.FromSlash("foo/user2")},
}}, }},
"work": {Root: "..", FileInfoPath: filepath.FromSlash("../work"), Nodes: map[string]Tree{ "work": {Root: "..", FileInfoPath: filepath.FromSlash("../work"), Nodes: map[string]tree{
"other": {Path: filepath.FromSlash("../work/other")}, "other": {Path: filepath.FromSlash("../work/other")},
}}, }},
}}, }},
}, },
{ {
targets: []string{"foo/user1", "../foo/other", "foo/user2"}, targets: []string{"foo/user1", "../foo/other", "foo/user2"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/user1")}, "user1": {Path: filepath.FromSlash("foo/user1")},
"user2": {Path: filepath.FromSlash("foo/user2")}, "user2": {Path: filepath.FromSlash("foo/user2")},
}}, }},
"foo-1": {Root: "..", FileInfoPath: filepath.FromSlash("../foo"), Nodes: map[string]Tree{ "foo-1": {Root: "..", FileInfoPath: filepath.FromSlash("../foo"), Nodes: map[string]tree{
"other": {Path: filepath.FromSlash("../foo/other")}, "other": {Path: filepath.FromSlash("../foo/other")},
}}, }},
}}, }},
@ -240,11 +240,11 @@ func TestTree(t *testing.T) {
}, },
}, },
targets: []string{"foo", "foo/work"}, targets: []string{"foo", "foo/work"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": { "foo": {
Root: ".", Root: ".",
FileInfoPath: "foo", FileInfoPath: "foo",
Nodes: map[string]Tree{ Nodes: map[string]tree{
"file": {Path: filepath.FromSlash("foo/file")}, "file": {Path: filepath.FromSlash("foo/file")},
"work": {Path: filepath.FromSlash("foo/work")}, "work": {Path: filepath.FromSlash("foo/work")},
}, },
@ -261,11 +261,11 @@ func TestTree(t *testing.T) {
}, },
}, },
targets: []string{"foo/work", "foo"}, targets: []string{"foo/work", "foo"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": { "foo": {
Root: ".", Root: ".",
FileInfoPath: "foo", FileInfoPath: "foo",
Nodes: map[string]Tree{ Nodes: map[string]tree{
"file": {Path: filepath.FromSlash("foo/file")}, "file": {Path: filepath.FromSlash("foo/file")},
"work": {Path: filepath.FromSlash("foo/work")}, "work": {Path: filepath.FromSlash("foo/work")},
}, },
@ -282,11 +282,11 @@ func TestTree(t *testing.T) {
}, },
}, },
targets: []string{"foo/work", "foo/work/user2"}, targets: []string{"foo/work", "foo/work/user2"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"work": { "work": {
FileInfoPath: filepath.FromSlash("foo/work"), FileInfoPath: filepath.FromSlash("foo/work"),
Nodes: map[string]Tree{ Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/work/user1")}, "user1": {Path: filepath.FromSlash("foo/work/user1")},
"user2": {Path: filepath.FromSlash("foo/work/user2")}, "user2": {Path: filepath.FromSlash("foo/work/user2")},
}, },
@ -304,10 +304,10 @@ func TestTree(t *testing.T) {
}, },
}, },
targets: []string{"foo/work/user2", "foo/work"}, targets: []string{"foo/work/user2", "foo/work"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"work": {FileInfoPath: filepath.FromSlash("foo/work"), "work": {FileInfoPath: filepath.FromSlash("foo/work"),
Nodes: map[string]Tree{ Nodes: map[string]tree{
"user1": {Path: filepath.FromSlash("foo/work/user1")}, "user1": {Path: filepath.FromSlash("foo/work/user1")},
"user2": {Path: filepath.FromSlash("foo/work/user2")}, "user2": {Path: filepath.FromSlash("foo/work/user2")},
}, },
@ -332,12 +332,12 @@ func TestTree(t *testing.T) {
}, },
}, },
targets: []string{"foo/work/user2/data/secret", "foo"}, targets: []string{"foo/work/user2/data/secret", "foo"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"other": {Path: filepath.FromSlash("foo/other")}, "other": {Path: filepath.FromSlash("foo/other")},
"work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]Tree{ "work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]tree{
"user2": {FileInfoPath: filepath.FromSlash("foo/work/user2"), Nodes: map[string]Tree{ "user2": {FileInfoPath: filepath.FromSlash("foo/work/user2"), Nodes: map[string]tree{
"data": {FileInfoPath: filepath.FromSlash("foo/work/user2/data"), Nodes: map[string]Tree{ "data": {FileInfoPath: filepath.FromSlash("foo/work/user2/data"), Nodes: map[string]tree{
"secret": { "secret": {
Path: filepath.FromSlash("foo/work/user2/data/secret"), Path: filepath.FromSlash("foo/work/user2/data/secret"),
}, },
@ -368,10 +368,10 @@ func TestTree(t *testing.T) {
}, },
unix: true, unix: true,
targets: []string{"mnt/driveA", "mnt/driveA/work/driveB"}, targets: []string{"mnt/driveA", "mnt/driveA/work/driveB"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"mnt": {Root: ".", FileInfoPath: filepath.FromSlash("mnt"), Nodes: map[string]Tree{ "mnt": {Root: ".", FileInfoPath: filepath.FromSlash("mnt"), Nodes: map[string]tree{
"driveA": {FileInfoPath: filepath.FromSlash("mnt/driveA"), Nodes: map[string]Tree{ "driveA": {FileInfoPath: filepath.FromSlash("mnt/driveA"), Nodes: map[string]tree{
"work": {FileInfoPath: filepath.FromSlash("mnt/driveA/work"), Nodes: map[string]Tree{ "work": {FileInfoPath: filepath.FromSlash("mnt/driveA/work"), Nodes: map[string]tree{
"driveB": { "driveB": {
Path: filepath.FromSlash("mnt/driveA/work/driveB"), Path: filepath.FromSlash("mnt/driveA/work/driveB"),
}, },
@ -384,9 +384,9 @@ func TestTree(t *testing.T) {
}, },
{ {
targets: []string{"foo/work/user", "foo/work/user"}, targets: []string{"foo/work/user", "foo/work/user"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]Tree{ "work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]tree{
"user": {Path: filepath.FromSlash("foo/work/user")}, "user": {Path: filepath.FromSlash("foo/work/user")},
}}, }},
}}, }},
@ -394,9 +394,9 @@ func TestTree(t *testing.T) {
}, },
{ {
targets: []string{"./foo/work/user", "foo/work/user"}, targets: []string{"./foo/work/user", "foo/work/user"},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]Tree{ "foo": {Root: ".", FileInfoPath: "foo", Nodes: map[string]tree{
"work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]Tree{ "work": {FileInfoPath: filepath.FromSlash("foo/work"), Nodes: map[string]tree{
"user": {Path: filepath.FromSlash("foo/work/user")}, "user": {Path: filepath.FromSlash("foo/work/user")},
}}, }},
}}, }},
@ -405,10 +405,10 @@ func TestTree(t *testing.T) {
{ {
win: true, win: true,
targets: []string{`c:\users\foobar\temp`}, targets: []string{`c:\users\foobar\temp`},
want: Tree{Nodes: map[string]Tree{ want: tree{Nodes: map[string]tree{
"c": {Root: `c:\`, FileInfoPath: `c:\`, Nodes: map[string]Tree{ "c": {Root: `c:\`, FileInfoPath: `c:\`, Nodes: map[string]tree{
"users": {FileInfoPath: `c:\users`, Nodes: map[string]Tree{ "users": {FileInfoPath: `c:\users`, Nodes: map[string]tree{
"foobar": {FileInfoPath: `c:\users\foobar`, Nodes: map[string]Tree{ "foobar": {FileInfoPath: `c:\users\foobar`, Nodes: map[string]tree{
"temp": {Path: `c:\users\foobar\temp`}, "temp": {Path: `c:\users\foobar\temp`},
}}, }},
}}, }},
@ -445,7 +445,7 @@ func TestTree(t *testing.T) {
back := rtest.Chdir(t, tempdir) back := rtest.Chdir(t, tempdir)
defer back() defer back()
tree, err := NewTree(fs.Local{}, test.targets) tree, err := newTree(fs.Local{}, test.targets)
if test.mustError { if test.mustError {
if err == nil { if err == nil {
t.Fatal("expected error, got nil") t.Fatal("expected error, got nil")