fix(download): separate torrent progress from transfer rates (#1654)

Co-authored-by: xunguangzlj-cloud <[email protected]>

pkg/download/downloader.go

@@ -345,7 +345,12 @@ func (d *Downloader) Setup() error {
 						}
 
 						current := task.fetcher.Progress().TotalDownloaded()
-						delta := current - task.Progress.Downloaded
+						delta := max(0, current-task.Progress.Downloaded)
+						if counter, ok := task.fetcher.(fetcher.DownloadCounter); ok && task.Status == base.DownloadStatusRunning {
+							received := counter.ReceivedBytes()
+							delta = received - task.receivedBytes
+							task.receivedBytes = received
+						}
 						if task.Meta != nil && task.Meta.Req != nil {
 							if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
 								downloaded, received := producer.Progress()
@@ -353,19 +358,20 @@ func (d *Downloader) Setup() error {
 									task.producer = producer
 									task.producerReceived = 0
 									task.speedArr = nil
+									task.downloadSampledAt = time.Time{}
 								}
 								current = downloaded
 								delta = max(0, received-task.producerReceived)
 								task.producerReceived = received
 							}
 						}
-						tick := float64(d.cfg.RefreshInterval) / 1000
 						downloadDataChanged := false
 						uploading := task.Uploading
 						if task.Status == base.DownloadStatusRunning {
 							downloadDataChanged = current != task.Progress.Downloaded
 							task.Progress.Used = task.timer.Used()
-							task.Progress.Speed = task.updateSpeed(max(0, delta), tick)
+							elapsed := sampleSeconds(&task.downloadSampledAt, time.Now())
+							task.Progress.Speed = task.updateSpeed(delta, elapsed)
 							task.Progress.Downloaded = current
 						}
 
@@ -374,7 +380,8 @@ func (d *Downloader) Setup() error {
 							uploader := task.fetcher.(fetcher.Uploader)
 							currentUploaded := uploader.UploadedBytes()
 							uploadDataChanged = currentUploaded != task.Progress.Uploaded
-							task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, tick)
+							elapsed := sampleSeconds(&task.uploadSampledAt, time.Now())
+							task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, elapsed)
 							task.Progress.Uploaded = currentUploaded
 						}
 						task.statusLock.Unlock()
@@ -2453,6 +2460,8 @@ func (d *Downloader) doStart(task *Task) (err error) {
 			return nil
 		}
 		task.Progress.Speed = 0
+		task.speedArr = nil
+		task.downloadSampledAt = time.Time{}
 		task.timer.Start()
 		task.statusLock.Unlock()
 		if err := task.fetcher.Start(); err != nil {
@@ -3061,8 +3070,10 @@ func initTask(task *Task) {
 	task.lock = &sync.Mutex{}
 	task.statsLock = &sync.Mutex{}
 	task.blobRefLock = &sync.Mutex{}
-	task.speedArr = make([]int64, 0)
-	task.uploadSpeedArr = make([]int64, 0)
+	task.speedArr = nil
+	task.uploadSpeedArr = nil
+	task.downloadSampledAt = time.Time{}
+	task.uploadSampledAt = time.Time{}
 }
 
 func (d *Downloader) releaseBlobTask(task *Task) {

pkg/download/model.go

@@ -75,20 +75,23 @@ type Task struct {
 	producer         production.Source
 	producerReceived int64
 
-	fetcherManager fetcher.FetcherManager
-	fetcher        fetcher.Fetcher
-	timer          *util.Timer
-	statusLock     *sync.Mutex
-	lock           *sync.Mutex
-	statsLock      *sync.Mutex
-	statsLoaded    bool
-	stats          *fetcher.Stats
-	statsSnapshot  json.RawMessage
-	blobRefLock    *sync.Mutex
-	blobURL        string
-	runGeneration  uint64
-	speedArr       []int64
-	uploadSpeedArr []int64
+	fetcherManager    fetcher.FetcherManager
+	fetcher           fetcher.Fetcher
+	timer             *util.Timer
+	statusLock        *sync.Mutex
+	lock              *sync.Mutex
+	statsLock         *sync.Mutex
+	statsLoaded       bool
+	stats             *fetcher.Stats
+	statsSnapshot     json.RawMessage
+	blobRefLock       *sync.Mutex
+	blobURL           string
+	runGeneration     uint64
+	speedArr          []speedSample
+	uploadSpeedArr    []speedSample
+	downloadSampledAt time.Time
+	uploadSampledAt   time.Time
+	receivedBytes     int64
 }
 
 // TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
@@ -195,7 +198,21 @@ func (t *Task) updateUploadSpeed(downloaded int64, usedTime float64) int64 {
 	return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime)
 }
 
-func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
+type speedSample struct {
+	bytes   int64
+	seconds float64
+}
+
+func sampleSeconds(lastSample *time.Time, now time.Time) float64 {
+	previous := *lastSample
+	*lastSample = now
+	if previous.IsZero() {
+		return 0
+	}
+	return now.Sub(previous).Seconds()
+}
+
+func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int64 {
 	if usedTime <= 0 {
 		return 0
 	}
@@ -204,18 +221,18 @@ func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
 		return 0
 	}
 
-	*speedArr = append(*speedArr, downloaded)
-	// Record last 5 seconds of download speed to calculate the average speed
-	if len(*speedArr) > int(5.0/usedTime) {
-		*speedArr = (*speedArr)[1:]
-	}
-
+	*speedArr = append(*speedArr, speedSample{bytes: downloaded, seconds: usedTime})
 	var total int64
-	for _, v := range *speedArr {
-		total += v
+	var elapsed float64
+	first := len(*speedArr)
+	for first > 0 && elapsed < 5 {
+		first--
+		sample := (*speedArr)[first]
+		total += sample.bytes
+		elapsed += sample.seconds
 	}
-
-	return int64(float64(total) / float64(len(*speedArr)) / usedTime)
+	*speedArr = (*speedArr)[first:]
+	return int64(float64(total) / elapsed)
 }
 
 type TaskFilter struct {

pkg/download/model_test.go

@@ -1,9 +1,32 @@
 package download
 
-import "testing"
+import (
+	"testing"
+	"time"
+)
+
+func TestTransferSpeedWithUnevenSamples(t *testing.T) {
+	var task Task
+	for _, sample := range []struct {
+		bytes   int64
+		seconds float64
+	}{
+		{1000, 1},
+		{2000, 2},
+		{4000, 4},
+		{6000, 6},
+	} {
+		if got := task.updateSpeed(sample.bytes, sample.seconds); got != 1000 {
+			t.Errorf("download speed after %v seconds = %d, want 1000", sample.seconds, got)
+		}
+		if got := task.updateUploadSpeed(sample.bytes, sample.seconds); got != 1000 {
+			t.Errorf("upload speed after %v seconds = %d, want 1000", sample.seconds, got)
+		}
+	}
+}
 
 func TestCalcSpeedResetOnRollback(t *testing.T) {
-	speedArr := []int64{1024, 2048, 4096}
+	speedArr := []speedSample{{1024, 1}, {2048, 1}, {4096, 1}}
 
 	if got := calcSpeed(&speedArr, -512, 1); got != 0 {
 		t.Fatalf("calcSpeed() = %d, want 0 after rollback", got)
@@ -16,3 +39,17 @@ func TestCalcSpeedResetOnRollback(t *testing.T) {
 		t.Fatalf("calcSpeed() = %d, want 1024 after reset", got)
 	}
 }
+
+func TestSampleSecondsUsesThePreviousSuccessfulSample(t *testing.T) {
+	start := time.Now()
+	var last time.Time
+	if got := sampleSeconds(&last, start); got != 0 {
+		t.Fatalf("first sample elapsed = %v, want 0", got)
+	}
+	if got := sampleSeconds(&last, start.Add(1500*time.Millisecond)); got != 1.5 {
+		t.Fatalf("elapsed after a delayed sample = %v, want 1.5", got)
+	}
+	if got := sampleSeconds(&last, start.Add(2*time.Second)); got != 0.5 {
+		t.Fatalf("elapsed after the next sample = %v, want 0.5", got)
+	}
+}