perf: write to disk only when progress updated (#839)

pkg/download/downloader.go

@@ -204,33 +204,29 @@ func (d *Downloader) Setup() error {
 
 						current := task.fetcher.Progress().TotalDownloaded()
 						tick := float64(d.cfg.RefreshInterval) / 1000
+						downloadDataChanged := false
 						if task.Status == base.DownloadStatusRunning {
+							downloadDataChanged = current != task.Progress.Downloaded
 							task.Progress.Used = task.timer.Used()
 							task.Progress.Speed = task.calcSpeed(task.speedArr, current-task.Progress.Downloaded, tick)
 							task.Progress.Downloaded = current
 						}
+
+						uploadDataChanged := false
 						if task.Uploading {
 							uploader := task.fetcher.(fetcher.Uploader)
 							currentUploaded := uploader.UploadedBytes()
+							uploadDataChanged = currentUploaded != task.Progress.Uploaded
 							task.Progress.UploadSpeed = task.calcSpeed(task.uploadSpeedArr, currentUploaded-task.Progress.Uploaded, tick)
 							task.Progress.Uploaded = currentUploaded
 						}
 						d.emit(EventKeyProgress, task)
 
-						// store fetcher progress
-						data, err := task.fetcherManager.Store(task.fetcher)
-						if err != nil {
-							d.Logger.Error().Stack().Err(err).Msgf("serialize fetcher failed: %s", task.ID)
-							return
-						}
-						if err := d.storage.Put(bucketSave, task.ID, data); err != nil {
-							d.Logger.Error().Stack().Err(err).Msgf("persist fetcher failed: %s", task.ID)
-							return
-						}
-						if err := d.storage.Put(bucketTask, task.ID, task); err != nil {
-							d.Logger.Error().Stack().Err(err).Msgf("persist task failed: %s", task.ID)
+						// store fetcher progress when download/upload data changed
+						if !downloadDataChanged && !uploadDataChanged {
 							return
 						}
+						d.saveTask(task)
 					}()
 				}
 			}
@@ -273,6 +269,23 @@ func (d *Downloader) setupFetcher(fm fetcher.FetcherManager, fetcher fetcher.Fet
 	fetcher.Setup(ctl)
 }
 
+func (d *Downloader) saveTask(task *Task) error {
+	data, err := task.fetcherManager.Store(task.fetcher)
+	if err != nil {
+		d.Logger.Error().Stack().Err(err).Msgf("serialize fetcher failed: %s", task.ID)
+		return err
+	}
+	if err := d.storage.Put(bucketSave, task.ID, data); err != nil {
+		d.Logger.Error().Stack().Err(err).Msgf("persist fetcher failed: %s", task.ID)
+		return err
+	}
+	if err := d.storage.Put(bucketTask, task.ID, task); err != nil {
+		d.Logger.Error().Stack().Err(err).Msgf("persist task failed: %s", task.ID)
+		return err
+	}
+	return nil
+}
+
 func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) {
 	rrId, err := gonanoid.New()
 	if err != nil {
@@ -1064,7 +1077,7 @@ func (d *Downloader) doStart(task *Task) (err error) {
 		if err := task.fetcher.Start(); err != nil {
 			return err
 		}
-		if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
+		if err := d.saveTask(task); err != nil {
 			return err
 		}
 		d.emit(EventKeyStart, task)