From 170bdeb944393ddd67ee4ffcca016ff0857d4dd1 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Sun, 11 Oct 2020 11:56:46 -0700 Subject: [PATCH] cli: improved restore progress information (#678) Demo: https://asciinema.org/a/8RV8SwBVt2CI0HSsklYwO8FhP --- cli/command_restore.go | 38 ++++++++++++++++++++-- snapshot/restore/restore.go | 64 +++++++++++++++++++++++++++++-------- 2 files changed, 85 insertions(+), 17 deletions(-) diff --git a/cli/command_restore.go b/cli/command_restore.go index 8257f0c0d..a878a4d1b 100644 --- a/cli/command_restore.go +++ b/cli/command_restore.go @@ -4,13 +4,16 @@ "archive/zip" "compress/gzip" "context" + "fmt" "os" "path/filepath" "strings" + "time" "github.com/pkg/errors" "gopkg.in/alecthomas/kingpin.v2" + "github.com/kopia/kopia/internal/clock" "github.com/kopia/kopia/internal/units" "github.com/kopia/kopia/repo" "github.com/kopia/kopia/snapshot/restore" @@ -43,6 +46,8 @@ 'kffbb7c28ea6c34d6cbe555d1cf80faa9' or 'kffbb7c28ea6c34d6cbe555d1cf80faa9/subdir1/subdir2' ` + + bitsPerByte = 8 ) var ( @@ -161,7 +166,7 @@ func detectRestoreMode(ctx context.Context, m string) string { } func printRestoreStats(ctx context.Context, st restore.Stats) { - log(ctx).Infof("Restored %v files, %v directories and %v symbolic links (%v)\n", st.FileCount, st.DirCount, st.SymlinkCount, units.BytesStringBase10(st.TotalFileSize)) + log(ctx).Infof("Restored %v files, %v directories and %v symbolic links (%v)\n", st.RestoredFileCount, st.RestoredDirCount, st.RestoredSymlinkCount, units.BytesStringBase10(st.RestoredTotalFileSize)) } func runRestoreCommand(ctx context.Context, rep repo.Repository) error { @@ -175,10 +180,37 @@ func runRestoreCommand(ctx context.Context, rep repo.Repository) error { return errors.Wrap(err, "unable to get filesystem entry") } + t0 := clock.Now() + st, err := restore.Entry(ctx, rep, output, rootEntry, restore.Options{ Parallel: restoreParallel, - ProgressCallback: func(ctx context.Context, enqueued, processing, completed int64) { - log(ctx).Infof("Restored %v/%v. Processing %v...", completed, enqueued, processing) + ProgressCallback: func(ctx context.Context, stats restore.Stats) { + restoredCount := stats.RestoredFileCount + stats.RestoredDirCount + stats.RestoredSymlinkCount + enqueuedCount := stats.EnqueuedFileCount + stats.EnqueuedDirCount + stats.EnqueuedSymlinkCount + + if restoredCount == 0 { + return + } + + var maybeRemaining string + + if stats.EnqueuedTotalFileSize > 0 { + progress := float64(stats.RestoredTotalFileSize) / float64(stats.EnqueuedTotalFileSize) + elapsed := clock.Since(t0) + if progress > 0 && elapsed.Seconds() > 1 { + predictedDuration := time.Duration(1e9 * elapsed.Seconds() / progress) + remaining := clock.Until(t0.Add(predictedDuration)).Truncate(time.Second) + bitsPerSecond := float64(stats.RestoredTotalFileSize) * bitsPerByte / elapsed.Seconds() + if remaining > time.Second { + maybeRemaining = fmt.Sprintf(" %v (%.1f%%) remaining %v", units.BitsPerSecondsString(bitsPerSecond), hundredPercent*progress, remaining) + } + } + } + + log(ctx).Infof("Processed %v (%v) of %v (%v)%v.", + restoredCount, units.BytesStringBase10(stats.RestoredTotalFileSize), + enqueuedCount, units.BytesStringBase10(stats.EnqueuedTotalFileSize), + maybeRemaining) }, }) if err != nil { diff --git a/snapshot/restore/restore.go b/snapshot/restore/restore.go index 451c04d6c..6c725c54d 100644 --- a/snapshot/restore/restore.go +++ b/snapshot/restore/restore.go @@ -28,25 +28,45 @@ type Output interface { // Stats represents restore statistics. type Stats struct { - TotalFileSize int64 - FileCount int32 - DirCount int32 - SymlinkCount int32 + RestoredTotalFileSize int64 + EnqueuedTotalFileSize int64 + + RestoredFileCount int32 + RestoredDirCount int32 + RestoredSymlinkCount int32 + EnqueuedFileCount int32 + EnqueuedDirCount int32 + EnqueuedSymlinkCount int32 +} + +func (s *Stats) clone() Stats { + return Stats{ + RestoredTotalFileSize: atomic.LoadInt64(&s.RestoredTotalFileSize), + EnqueuedTotalFileSize: atomic.LoadInt64(&s.EnqueuedTotalFileSize), + RestoredFileCount: atomic.LoadInt32(&s.RestoredFileCount), + RestoredDirCount: atomic.LoadInt32(&s.RestoredDirCount), + RestoredSymlinkCount: atomic.LoadInt32(&s.RestoredSymlinkCount), + EnqueuedFileCount: atomic.LoadInt32(&s.EnqueuedFileCount), + EnqueuedDirCount: atomic.LoadInt32(&s.EnqueuedDirCount), + EnqueuedSymlinkCount: atomic.LoadInt32(&s.EnqueuedSymlinkCount), + } } // Options provides optional restore parameters. type Options struct { Parallel int - ProgressCallback func(ctx context.Context, enqueued, active, completed int64) + ProgressCallback func(ctx context.Context, s Stats) } // Entry walks a snapshot root with given root entry and restores it to the provided output. func Entry(ctx context.Context, rep repo.Repository, output Output, rootEntry fs.Entry, options Options) (Stats, error) { c := copier{output: output, q: parallelwork.NewQueue()} - c.q.ProgressCallback = options.ProgressCallback + c.q.ProgressCallback = func(ctx context.Context, enqueued, active, completed int64) { + options.ProgressCallback(ctx, c.stats.clone()) + } - c.q.EnqueueBack(ctx, func() error { + c.q.EnqueueFront(ctx, func() error { return errors.Wrap(c.copyEntry(ctx, rootEntry, "", func() error { return nil }), "error copying") }) @@ -84,8 +104,8 @@ func (c *copier) copyEntry(ctx context.Context, e fs.Entry, targetPath string, o case fs.File: log(ctx).Debugf("file: '%v'", targetPath) - atomic.AddInt32(&c.stats.FileCount, 1) - atomic.AddInt64(&c.stats.TotalFileSize, e.Size()) + atomic.AddInt32(&c.stats.RestoredFileCount, 1) + atomic.AddInt64(&c.stats.RestoredTotalFileSize, e.Size()) if err := c.output.WriteFile(ctx, targetPath, e); err != nil { return errors.Wrap(err, "copy file") @@ -94,7 +114,7 @@ func (c *copier) copyEntry(ctx context.Context, e fs.Entry, targetPath string, o return onCompletion() case fs.Symlink: - atomic.AddInt32(&c.stats.SymlinkCount, 1) + atomic.AddInt32(&c.stats.RestoredSymlinkCount, 1) log(ctx).Debugf("symlink: '%v'", targetPath) if err := c.output.CreateSymlink(ctx, targetPath, e); err != nil { @@ -109,7 +129,7 @@ func (c *copier) copyEntry(ctx context.Context, e fs.Entry, targetPath string, o } func (c *copier) copyDirectory(ctx context.Context, d fs.Directory, targetPath string, onCompletion parallelwork.CallbackFunc) error { - atomic.AddInt32(&c.stats.DirCount, 1) + atomic.AddInt32(&c.stats.RestoredDirCount, 1) if err := c.output.BeginDirectory(ctx, targetPath, d); err != nil { return errors.Wrap(err, "create directory") @@ -139,9 +159,25 @@ func (c *copier) copyDirectoryContent(ctx context.Context, d fs.Directory, targe for _, e := range entries { e := e - c.q.EnqueueBack(ctx, func() error { - return c.copyEntry(ctx, e, path.Join(targetPath, e.Name()), onItemCompletion) - }) + if e.IsDir() { + atomic.AddInt32(&c.stats.EnqueuedDirCount, 1) + // enqueue directories first, so that we quickly determine the total number and size of items. + c.q.EnqueueFront(ctx, func() error { + return c.copyEntry(ctx, e, path.Join(targetPath, e.Name()), onItemCompletion) + }) + } else { + if isSymlink(e) { + atomic.AddInt32(&c.stats.EnqueuedSymlinkCount, 1) + } else { + atomic.AddInt32(&c.stats.EnqueuedFileCount, 1) + } + + atomic.AddInt64(&c.stats.EnqueuedTotalFileSize, e.Size()) + + c.q.EnqueueBack(ctx, func() error { + return c.copyEntry(ctx, e, path.Join(targetPath, e.Name()), onItemCompletion) + }) + } } return nil