Send fetch progress at most once per second (closes #206)
check / check (push) Waiting to run
check / check (push) Waiting to run
progressWriter sent an update for every chunk io.Copy wrote, so fetch logged a line per 32 KiB. It now sends at most one per second, waiting a second before the first as gen and check do, and one more when the last byte is written. TestFetchProgress now checks that a download of several chunks sends no more than one update per second plus the last. Model: opus-5-5
This commit is contained in:
+24
-14
@@ -679,21 +679,26 @@ func resolveManifestURL(inputURL string) (string, error) {
|
||||
return parsed.JoinPath(defaultManifestName).String(), nil
|
||||
}
|
||||
|
||||
// progressWriter wraps an io.Writer and reports progress to a channel.
|
||||
// progressWriter wraps an io.Writer and reports progress to a channel,
|
||||
// at most once per second and once more when the last byte is written.
|
||||
type progressWriter struct {
|
||||
w io.Writer
|
||||
path string
|
||||
total int64
|
||||
written int64
|
||||
startTime time.Time
|
||||
progress chan<- DownloadProgress
|
||||
w io.Writer
|
||||
path string
|
||||
total int64
|
||||
written int64
|
||||
startTime time.Time
|
||||
lastProgress time.Time
|
||||
progress chan<- DownloadProgress
|
||||
}
|
||||
|
||||
func (pw *progressWriter) Write(p []byte) (int, error) {
|
||||
n, err := pw.w.Write(p)
|
||||
|
||||
pw.written += int64(n)
|
||||
if pw.progress != nil {
|
||||
|
||||
isLast := pw.written == pw.total
|
||||
if pw.progress != nil &&
|
||||
(isLast || time.Since(pw.lastProgress) >= time.Second) {
|
||||
var (
|
||||
bytesPerSec float64
|
||||
eta time.Duration
|
||||
@@ -715,6 +720,8 @@ func (pw *progressWriter) Write(p []byte) (int, error) {
|
||||
BytesPerSec: bytesPerSec,
|
||||
ETA: eta,
|
||||
})
|
||||
|
||||
pw.lastProgress = time.Now()
|
||||
}
|
||||
|
||||
return n, err
|
||||
@@ -899,13 +906,16 @@ func saveResponse(
|
||||
// Set up hash computation
|
||||
h := sha256.New()
|
||||
|
||||
// Create progress-reporting writer that also computes hash
|
||||
// Create progress-reporting writer that also computes hash. The first
|
||||
// update waits a second, as gen's and check's do.
|
||||
start := time.Now()
|
||||
pw := &progressWriter{
|
||||
w: io.MultiWriter(out, h),
|
||||
path: localPath,
|
||||
total: totalBytes,
|
||||
startTime: time.Now(),
|
||||
progress: progress,
|
||||
w: io.MultiWriter(out, h),
|
||||
path: localPath,
|
||||
total: totalBytes,
|
||||
startTime: start,
|
||||
lastProgress: start,
|
||||
progress: progress,
|
||||
}
|
||||
|
||||
// Copy content while hashing and reporting progress. One byte past
|
||||
|
||||
@@ -516,7 +516,7 @@ func TestManifestDownloadStopsAtLimit(t *testing.T) {
|
||||
func TestFetchProgress(t *testing.T) {
|
||||
// Create source filesystem with a larger test file
|
||||
sourceFs := afero.NewMemMapFs()
|
||||
// Create content large enough to trigger multiple progress updates
|
||||
// Content large enough that io.Copy writes it in at least four chunks
|
||||
content := bytes.Repeat([]byte("x"), 100*1024) // 100KB
|
||||
require.NoError(t, afero.WriteFile(sourceFs, "/large.txt", content, 0o644))
|
||||
|
||||
@@ -559,8 +559,11 @@ func TestFetchProgress(t *testing.T) {
|
||||
}()
|
||||
|
||||
// Download
|
||||
start := time.Now()
|
||||
err = downloadFile(context.Background(), testClient(),
|
||||
server.URL+"/large.txt", ".", "large.txt", files[0], progress)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
close(progress)
|
||||
<-done
|
||||
|
||||
@@ -569,6 +572,12 @@ func TestFetchProgress(t *testing.T) {
|
||||
// Verify we got progress updates
|
||||
assert.NotEmpty(t, progressUpdates, "should have received progress updates")
|
||||
|
||||
// At most one update per second, plus the one reporting the file
|
||||
// complete: a download that takes under a second sends two at most,
|
||||
// however many chunks it writes.
|
||||
assert.LessOrEqual(t, len(progressUpdates), 2+int(elapsed/time.Second),
|
||||
"one progress update per second at most, plus the last")
|
||||
|
||||
// Verify final progress shows complete
|
||||
if len(progressUpdates) > 0 {
|
||||
last := progressUpdates[len(progressUpdates)-1]
|
||||
|
||||
Reference in New Issue
Block a user