Speed monitor

Author: colin969

downloader.go

@@ -14,38 +14,43 @@ import (
 )
 
 type Update struct {
-	IndexFile *IndexedFile
-	Progress  float64
-	Done      bool
+	IndexFile       *IndexedFile
+	Retry           bool
+	RemoveTakenFlag bool
+	Progress        float64
+	Bytes           int64
+	Done            bool
 }
 
 type Downloader struct {
-	RateLimit   *int
-	state       *InstallerState
-	ctx         context.Context
-	cancel      context.CancelFunc
-	client      *grab.Client
-	reqch       chan *grab.Request
-	respch      chan *grab.Response
-	updatech    chan *Update
-	workerWg    sync.WaitGroup
-	responderWg sync.WaitGroup
-	updaterWg   sync.WaitGroup
-	running     bool
-	started     bool
-	installPath string
+	RateLimit    *int
+	state        *InstallerState
+	ctx          context.Context
+	cancel       context.CancelFunc
+	client       *grab.Client
+	reqch        chan *grab.Request
+	respch       chan *grab.Response
+	updatech     chan *Update
+	workerWg     sync.WaitGroup
+	responderWg  sync.WaitGroup
+	updaterWg    sync.WaitGroup
+	newRequestWg sync.WaitGroup
+	running      bool
+	started      bool
+	installPath  string
 }
 
 func NewDownloader(state *InstallerState) *Downloader {
 	d := &Downloader{
-		RateLimit:   nil,
-		state:       state,
-		client:      grab.NewClient(),
-		workerWg:    sync.WaitGroup{},
-		responderWg: sync.WaitGroup{},
-		updaterWg:   sync.WaitGroup{},
-		running:     false,
-		started:     false,
+		RateLimit:    nil,
+		state:        state,
+		client:       grab.NewClient(),
+		workerWg:     sync.WaitGroup{},
+		responderWg:  sync.WaitGroup{},
+		updaterWg:    sync.WaitGroup{},
+		newRequestWg: sync.WaitGroup{},
+		running:      false,
+		started:      false,
 	}
 
 	return d
@@ -97,21 +102,48 @@ func (d *Downloader) Resume() error {
 						d.updatech <- &Update{
 							IndexFile: f,
 							Progress:  resp.Progress(),
+							Bytes:     resp.BytesComplete(),
 							Done:      false,
 						}
 					case <-resp.Done:
 						// Done, check for error
+						f := resp.Request.Tag.(*IndexedFile)
 						err := resp.Err()
 						if err != nil {
 							fmt.Println(err.Error())
-							// Bad download
+							if err.Error() == "context canceled" {
+								d.updatech <- &Update{
+									IndexFile:       f,
+									Retry:           false,
+									RemoveTakenFlag: true,
+									Progress:        1,
+									Bytes:           0,
+									Done:            true,
+								}
+								return
+							} else {
+								// Bad download, retry if below 3 retries
+								if f.RetryCount < 3 {
+									f.RetryCount += 1
+									d.updatech <- &Update{
+										IndexFile:       f,
+										Retry:           true,
+										RemoveTakenFlag: false,
+										Progress:        1,
+										Bytes:           0,
+										Done:            true,
+									}
+								}
+							}
 						} else {
 							// Successful download, notify UI updater
-							f := resp.Request.Tag.(*IndexedFile)
 							d.updatech <- &Update{
-								IndexFile: f,
-								Progress:  1,
-								Done:      true,
+								IndexFile:       f,
+								Retry:           false,
+								RemoveTakenFlag: false,
+								Progress:        1,
+								Bytes:           resp.BytesComplete(),
+								Done:            true,
 							}
 						}
 						return
@@ -127,6 +159,7 @@ func (d *Downloader) Resume() error {
 	d.updaterWg.Add(1)
 	go func() {
 		defer d.updaterWg.Done()
+
 		// Create UI files
 		uifiles := make([]*UiFile, 4)
 		for i := 0; i < 4; i++ {
@@ -137,14 +170,117 @@ func (d *Downloader) Resume() error {
 			}
 		}
 
+		// Create speed handler
+		speedch := make(chan int64, 4)
+		defer close(speedch)
+		// Set up download speed handler to store records and update average
+		d.updaterWg.Add(1)
+		go func() {
+			defer d.updaterWg.Done()
+			// Set UI speed value to 0 when downloader stops
+			defer func() { _ = d.state.formatDownloadSpeed.Set("0.0B/s") }()
+			// Track 6 speed records to get average
+			bytePerSecondRecords := make([]float64, 0)
+
+			queue := func(bytesPerSecond float64) {
+				if len(bytePerSecondRecords) > 6 {
+					// Remove first element
+					bytePerSecondRecords = bytePerSecondRecords[1:]
+				}
+				// Add new record
+				bytePerSecondRecords = append(bytePerSecondRecords, bytesPerSecond)
+			}
+
+			averageSpeed := func() float64 {
+				if len(bytePerSecondRecords) == 0 {
+					return 0
+				}
+
+				sum := float64(0)
+				for _, num := range bytePerSecondRecords {
+					sum += num
+				}
+
+				return sum / float64(len(bytePerSecondRecords))
+			}
+
+			totalBytes := int64(0)
+			lastByteRecord := int64(0)
+			lastUpdate := time.Now()
+
+			for bytes := range speedch {
+				totalBytes += bytes
+
+				curTime := time.Now()
+				secondsDiff := curTime.Sub(lastUpdate).Seconds()
+				if secondsDiff > 0.5 {
+					byteDiff := totalBytes - lastByteRecord
+					lastByteRecord = totalBytes
+					lastUpdate = curTime
+					// Push record
+					queue(float64(byteDiff) / secondsDiff)
+					// Update UI
+					d.state.downloadSpeed = averageSpeed()
+					_ = d.state.formatDownloadSpeed.Set(FormatBytes(int64(d.state.downloadSpeed)) + "/s")
+				}
+			}
+		}()
+
+		_ = d.state.fileTitle1.Set("None")
+		_ = d.state.fileProgress1.Set(0)
+		_ = d.state.fileTitle2.Set("None")
+		_ = d.state.fileProgress2.Set(0)
+		_ = d.state.fileTitle3.Set("None")
+		_ = d.state.fileProgress3.Set(0)
+		_ = d.state.fileTitle4.Set("None")
+		_ = d.state.fileProgress4.Set(0)
+
 		for update := range d.updatech {
-			fmt.Println(fmt.Sprintf("path: %s, progress: %f", update.IndexFile.Filepath, update.Progress))
+			// Failed download because of context cancel, remove taken flag instead, ignore ui update
+			if update.RemoveTakenFlag {
+				err = d.state.Repo.ClearTaken(update.IndexFile)
+				if err != nil {
+					dialog.NewError(&DatabaseError{err}, d.state.window).Show()
+				}
+				continue
+			}
+
+			// Immediately retry file if asked, ignore ui update
+			if update.Retry {
+				d.newRequestWg.Add(1)
+				go func() {
+					defer d.newRequestWg.Done()
+					select {
+					case <-d.ctx.Done():
+						{
+							// Kill request channel since we are sole sender
+							close(d.reqch)
+							return
+						}
+					default:
+						{
+							req, err := d.NewRequest(update.IndexFile)
+							if err != nil {
+								dialog.NewError(err, d.state.window).Show()
+							}
+							d.reqch <- req
+						}
+					}
+				}()
+				continue
+			}
+
 			// Update UI element
 			updateIdx := -1
 			for idx, f := range uifiles {
 				if f.Filepath == update.IndexFile.Filepath {
+					// Send bytes update to speed handler
+					bytesDiff := f.Bytes - update.Bytes
+					speedch <- bytesDiff
+					// Update ui file
 					f.Progress = update.Progress
 					f.Done = update.Done
+					f.Bytes = update.Bytes
 					updateIdx = idx
 					break
 				}
@@ -153,9 +289,14 @@ func (d *Downloader) Resume() error {
 				// Didn't find existing entry, find an older one to replace
 				for idx, f := range uifiles {
 					if f.Done == true {
+						// Send bytes update to speed handler
+						bytesDiff := f.Bytes
+						speedch <- bytesDiff
+						// Update ui file
 						f.Filepath = update.IndexFile.Filepath
 						f.Progress = update.Progress
 						f.Done = update.Done
+						f.Bytes = update.Bytes
 						updateIdx = idx
 						break
 					}
@@ -179,7 +320,6 @@ func (d *Downloader) Resume() error {
 					_ = d.state.fileTitle4.Set(update.IndexFile.Filepath)
 					_ = d.state.fileProgress4.Set(update.Progress)
 				}
-
 			}
 
 			if update.Done {
@@ -189,13 +329,13 @@ func (d *Downloader) Resume() error {
 					dialog.NewError(&DatabaseError{err}, d.state.window).Show()
 				}
 
-				// Add new request if still running
+				d.newRequestWg.Add(1)
 				go func() {
+					defer d.newRequestWg.Done()
 					select {
 					case <-d.ctx.Done():
 						{
-							// Kill request channel since we are sole sender
-							close(d.reqch)
+							// Context dead, assume channel closed already by parent
 							return
 						}
 					default:
@@ -225,7 +365,6 @@ func (d *Downloader) Resume() error {
 					dialog.NewError(err, d.state.window).Show()
 				}
 				progress := float64(d.state.downloadedSize) / float64(d.state.totalSize)
-				fmt.Println(progress)
 				err = d.state.progressBarTotal.Set(progress)
 				if err != nil {
 					dialog.NewError(err, d.state.window).Show()
@@ -262,8 +401,10 @@ func (d *Downloader) Stop() {
 
 	// Stop new and current requests
 	d.cancel()
+	d.newRequestWg.Wait()
+	close(d.reqch)
 
-	// Wait for all workers to exit
+	// Let workers try to trigger exit first
 	d.workerWg.Wait()
 	close(d.respch)
 

index.py

@@ -1,7 +1,5 @@
 #!/usr/bin/env python3
 from tqdm import tqdm
-import json
-import lzma
 import os
 import sys
 import hashlib

main.go

@@ -89,7 +89,7 @@ func setupLayout(w fyne.Window, state *InstallerState) *fyne.Container {
 			if f == nil {
 				return
 			}
-			fmt.Printf("chosen: %v", f.Path())
+			fmt.Printf("chosen: %v\n", f.Path())
 			// Validate path
 			p, resumable, err := validatePath(f.Path())
 			if err != nil {
@@ -124,7 +124,6 @@ func setupLayout(w fyne.Window, state *InstallerState) *fyne.Container {
 				_ = state.formatDownloadedSize.Set(FormatBytes(state.downloadedSize))
 				_ = state.formatTotalSize.Set(FormatBytes(state.totalSize))
 				progress := float64(state.downloadedSize) / float64(state.totalSize) * 100
-				fmt.Println(progress)
 				err = state.progressBarTotal.Set(progress)
 				if err != nil {
 					dialog.NewError(err, w).Show()
@@ -193,13 +192,13 @@ func mainLayout(w fyne.Window, state *InstallerState) *fyne.Container {
 		fileHeader2)
 	fileProgressBar2 := widget.NewProgressBarWithData(state.fileProgress2)
 
-	fileHeader3 := widget.NewLabelWithData(state.fileTitle1)
+	fileHeader3 := widget.NewLabelWithData(state.fileTitle3)
 	fileContainer3 := container.New(layout.NewHBoxLayout(),
 		fileLabel,
 		fileHeader3)
 	fileProgressBar3 := widget.NewProgressBarWithData(state.fileProgress3)
 
-	fileHeader4 := widget.NewLabelWithData(state.fileTitle1)
+	fileHeader4 := widget.NewLabelWithData(state.fileTitle4)
 	fileContainer4 := container.New(layout.NewHBoxLayout(),
 		fileLabel,
 		fileHeader4)
@@ -208,6 +207,9 @@ func mainLayout(w fyne.Window, state *InstallerState) *fyne.Container {
 	progressBarTotal := widget.NewProgressBarWithData(state.progressBarTotal)
 	totalLabel := canvas.NewText("Total Progress...", color.White)
 
+	//rateLimitLabel := widget.NewLabel("Download Speed Limit:")
+	//rateLimitEntry := widget.NewEntry()
+
 	// Create buttons
 	button1 := widget.NewButton("Start", func() {
 		err := state.Grabber.Resume()
@@ -232,6 +234,11 @@ func mainLayout(w fyne.Window, state *InstallerState) *fyne.Container {
 			return
 		}
 
+		// Update progress state
+		_ = state.progressBarTotal.Set(0)
+		state.downloadedSize = 0
+		_ = state.formatDownloadedSize.Set("0.0B")
+
 		// Start downloader again
 		err = state.Grabber.Resume()
 		if err != nil {
@@ -259,7 +266,7 @@ func mainLayout(w fyne.Window, state *InstallerState) *fyne.Container {
 		widget.NewLabel("/"),
 		totalSizeLabel)
 
-	speedHeaderLabel := widget.NewLabel("Current Download Speed:")
+	speedHeaderLabel := widget.NewLabel("Average Speed:")
 	speedHeaderLabel.Alignment = fyne.TextAlignLeading
 	speedLabel := widget.NewLabelWithData(state.formatDownloadSpeed)
 	speedLabel.Alignment = fyne.TextAlignLeading