From cfdbc407564d0d07b635a2cf73810c8f78350b9e Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Mon, 26 Jul 2021 18:57:58 -0700 Subject: [PATCH] cli: Fixed ETA estimation of 'snapshot verify' (#1213) --- cli/command_snapshot_verify.go | 4 ++-- internal/parallelwork/parallel_work_queue.go | 12 ++++++++---- internal/timetrack/estimator.go | 2 +- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/cli/command_snapshot_verify.go b/cli/command_snapshot_verify.go index 8b0de0275..00ce23247 100644 --- a/cli/command_snapshot_verify.go +++ b/cli/command_snapshot_verify.go @@ -61,8 +61,8 @@ type verifier struct { func (v *verifier) progressCallback(ctx context.Context, enqueued, active, completed int64) { maybeTimeRemaining := "" - if est, ok := v.tt.Estimate(float64(completed), float64(enqueued)+float64(completed)); ok { - maybeTimeRemaining = fmt.Sprintf(" remaining %v (ETA %v)", est.Remaining, formatTimestamp(est.EstimatedEndTime)) + if est, ok := v.tt.Estimate(float64(completed), float64(enqueued)); ok { + maybeTimeRemaining = fmt.Sprintf(", %.1f%% complete, remaining %v (ETA %v)", est.PercentComplete, est.Remaining, formatTimestamp(est.EstimatedEndTime)) } log(ctx).Infof("Found %v objects, verifying %v, completed %v objects%v.", enqueued, active, completed, maybeTimeRemaining) diff --git a/internal/parallelwork/parallel_work_queue.go b/internal/parallelwork/parallel_work_queue.go index 443a0a12a..5e264d178 100644 --- a/internal/parallelwork/parallel_work_queue.go +++ b/internal/parallelwork/parallel_work_queue.go @@ -59,6 +59,8 @@ func (v *Queue) enqueue(ctx context.Context, front bool, callback CallbackFunc) // Process starts N workers, which will be processing elements in the queue until the queue // is empty and all workers are idle or until any of the workers returns an error. func (v *Queue) Process(ctx context.Context, workers int) error { + defer v.reportProgress(ctx) + eg, ctx := errgroup.WithContext(context.Background()) for i := 0; i < workers; i++ { @@ -125,19 +127,21 @@ func (v *Queue) completed(ctx context.Context) { v.monitor.Broadcast() } -func (v *Queue) maybeReportProgress(ctx context.Context) { +func (v *Queue) reportProgress(ctx context.Context) { cb := v.ProgressCallback - if cb == nil { - return + if cb != nil { + cb(ctx, v.enqueuedWork, v.activeWorkerCount, v.completedWork) } +} +func (v *Queue) maybeReportProgress(ctx context.Context) { if clock.Now().Before(v.nextReportTime) { return } v.nextReportTime = clock.Now().Add(1 * time.Second) - cb(ctx, v.enqueuedWork, v.activeWorkerCount, v.completedWork) + v.reportProgress(ctx) } // OnNthCompletion invokes the provided callback once the returned callback function has been invoked exactly n times. diff --git a/internal/timetrack/estimator.go b/internal/timetrack/estimator.go index 1872a2918..962aabb00 100644 --- a/internal/timetrack/estimator.go +++ b/internal/timetrack/estimator.go @@ -29,7 +29,7 @@ func (v Estimator) Estimate(completed, total float64) (Timings, bool) { if elapsed > 1*time.Second && total > 0 && completed > 0 { completedRatio := completed / total if completedRatio > 1 { - completedRatio = 0 + completedRatio = 1 } if completedRatio < 0 {