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 was merged in pull request #215.
This commit is contained in:
+24
-14
@@ -679,21 +679,26 @@ func resolveManifestURL(inputURL string) (string, error) {
|
|||||||
return parsed.JoinPath(defaultManifestName).String(), nil
|
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 {
|
type progressWriter struct {
|
||||||
w io.Writer
|
w io.Writer
|
||||||
path string
|
path string
|
||||||
total int64
|
total int64
|
||||||
written int64
|
written int64
|
||||||
startTime time.Time
|
startTime time.Time
|
||||||
progress chan<- DownloadProgress
|
lastProgress time.Time
|
||||||
|
progress chan<- DownloadProgress
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pw *progressWriter) Write(p []byte) (int, error) {
|
func (pw *progressWriter) Write(p []byte) (int, error) {
|
||||||
n, err := pw.w.Write(p)
|
n, err := pw.w.Write(p)
|
||||||
|
|
||||||
pw.written += int64(n)
|
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 (
|
var (
|
||||||
bytesPerSec float64
|
bytesPerSec float64
|
||||||
eta time.Duration
|
eta time.Duration
|
||||||
@@ -715,6 +720,8 @@ func (pw *progressWriter) Write(p []byte) (int, error) {
|
|||||||
BytesPerSec: bytesPerSec,
|
BytesPerSec: bytesPerSec,
|
||||||
ETA: eta,
|
ETA: eta,
|
||||||
})
|
})
|
||||||
|
|
||||||
|
pw.lastProgress = time.Now()
|
||||||
}
|
}
|
||||||
|
|
||||||
return n, err
|
return n, err
|
||||||
@@ -899,13 +906,16 @@ func saveResponse(
|
|||||||
// Set up hash computation
|
// Set up hash computation
|
||||||
h := sha256.New()
|
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{
|
pw := &progressWriter{
|
||||||
w: io.MultiWriter(out, h),
|
w: io.MultiWriter(out, h),
|
||||||
path: localPath,
|
path: localPath,
|
||||||
total: totalBytes,
|
total: totalBytes,
|
||||||
startTime: time.Now(),
|
startTime: start,
|
||||||
progress: progress,
|
lastProgress: start,
|
||||||
|
progress: progress,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Copy content while hashing and reporting progress. One byte past
|
// Copy content while hashing and reporting progress. One byte past
|
||||||
|
|||||||
@@ -530,7 +530,7 @@ func TestManifestDownloadStopsAtLimit(t *testing.T) {
|
|||||||
func TestFetchProgress(t *testing.T) {
|
func TestFetchProgress(t *testing.T) {
|
||||||
// Create source filesystem with a larger test file
|
// Create source filesystem with a larger test file
|
||||||
sourceFs := afero.NewMemMapFs()
|
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
|
content := bytes.Repeat([]byte("x"), 100*1024) // 100KB
|
||||||
require.NoError(t, afero.WriteFile(sourceFs, "/large.txt", content, 0o644))
|
require.NoError(t, afero.WriteFile(sourceFs, "/large.txt", content, 0o644))
|
||||||
|
|
||||||
@@ -576,8 +576,11 @@ func TestFetchProgress(t *testing.T) {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
// Download
|
// Download
|
||||||
|
start := time.Now()
|
||||||
err = downloadFile(context.Background(), testClient(),
|
err = downloadFile(context.Background(), testClient(),
|
||||||
server.URL+"/large.txt", ".", "large.txt", files[0], progress)
|
server.URL+"/large.txt", ".", "large.txt", files[0], progress)
|
||||||
|
elapsed := time.Since(start)
|
||||||
|
|
||||||
close(progress)
|
close(progress)
|
||||||
<-done
|
<-done
|
||||||
|
|
||||||
@@ -586,6 +589,12 @@ func TestFetchProgress(t *testing.T) {
|
|||||||
// Verify we got progress updates
|
// Verify we got progress updates
|
||||||
assert.NotEmpty(t, progressUpdates, "should have received 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
|
// Verify final progress shows complete
|
||||||
if len(progressUpdates) > 0 {
|
if len(progressUpdates) > 0 {
|
||||||
last := progressUpdates[len(progressUpdates)-1]
|
last := progressUpdates[len(progressUpdates)-1]
|
||||||
|
|||||||
Reference in New Issue
Block a user