change: improve measurement algorithm with dynamic rate

.github/workflows/ci.yaml

@@ -9,7 +9,7 @@ jobs:
       - name: set up
         uses: actions/setup-go@v3
         with:
-          go-version: ^1.16
+          go-version: ^1.18
         id: go
       - name: check out
         uses: actions/checkout@v3
@@ -43,9 +43,9 @@ jobs:
     steps:
       - uses: actions/setup-go@v3
         with:
-          go-version: 1.16.5
+          go-version: 1.18
       - uses: actions/checkout@v3
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3
         with:
-          version: v1.29
+          version: v1.51.1

.github/workflows/release.yml

@@ -18,7 +18,7 @@ jobs:
         name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
       -
         name: Run GoReleaser
         uses: goreleaser/goreleaser-action@v2

go.mod

@@ -1,10 +1,13 @@
 module github.com/showwin/speedtest-go
 
-go 1.16
+go 1.18
+
+require (
+	github.com/LyricTian/queue v1.3.0
+	gopkg.in/alecthomas/kingpin.v2 v2.2.6
+)
 
 require (
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect
 	github.com/alecthomas/units v0.0.0-20210208195552-ff826a37aa15 // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	gopkg.in/alecthomas/kingpin.v2 v2.2.6
 )

go.sum

@@ -1,18 +1,18 @@
+github.com/LyricTian/queue v1.3.0 h1:1xEFZlteW6iu5Qbrz7ZsiSKMKaxY1bQHsbx0jrB1pDA=
+github.com/LyricTian/queue v1.3.0/go.mod h1:pbkoplz/zRToCay3pRjz75P8fQAgvkRKJdEzVUQYhXY=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
-github.com/alecthomas/units v0.0.0-20201120081800-1786d5ef83d4 h1:EBTWhcAX7rNQ80RLwLCpHZBBrJuzallFHnF+yMXo928=
-github.com/alecthomas/units v0.0.0-20201120081800-1786d5ef83d4/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
 github.com/alecthomas/units v0.0.0-20210208195552-ff826a37aa15 h1:AUNCr9CiJuwrRYS3XieqF+Z9B9gNxo/eANAJCF2eiN4=
 github.com/alecthomas/units v0.0.0-20210208195552-ff826a37aa15/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
-golang.org/x/sync v0.0.0-20201207232520-09787c993a3a h1:DcqTD9SDLc+1P/r1EmRBwnVsrOwW+kk2vWf9n+1sGhs=
-golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

speedtest.go

@@ -97,6 +97,8 @@ func main() {
 
 func startTest(servers speedtest.Servers, savingMode bool, jsonOutput bool) {
 	for _, s := range servers {
+		// Reset DataManager counters, avoid measurement of multiple server result mixing.
+		speedtest.GlobalDataManager.Reset()
 		if !jsonOutput {
 			showServer(s)
 		}
@@ -107,7 +109,7 @@ func startTest(servers speedtest.Servers, savingMode bool, jsonOutput bool) {
 		if jsonOutput {
 			err := s.DownloadTest(savingMode)
 			checkError(err)
-
+			time.Sleep(time.Second * 5)
 			err = s.UploadTest(savingMode)
 			checkError(err)
 
@@ -118,9 +120,11 @@ func startTest(servers speedtest.Servers, savingMode bool, jsonOutput bool) {
 
 		err = testDownload(s, savingMode)
 		checkError(err)
+		// It is necessary to wait for the release of the last test resource,
+		// otherwise the overload will cause excessive data deviation
+		time.Sleep(time.Second * 5)
 		err = testUpload(s, savingMode)
 		checkError(err)
-
 		showServerResult(s)
 	}
 

speedtest/data_manager.go

@@ -0,0 +1,191 @@
+package speedtest
+
+import (
+	"bytes"
+	"io"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const readChunkSize = 1024 * 32 // 32 KBytes
+const rateCaptureFrequency = time.Second
+
+type DataType int32
+
+const TypeDownload = 0
+const TypeUpload = 1
+
+type DataManager struct {
+	totalDownload int64
+	totalUpload   int64
+
+	DownloadRateSequence []float64
+	UploadRateSequence   []float64
+
+	DataGroup []*DataChunk
+	sync.Mutex
+
+	repeatByte *[]byte
+}
+
+func NewDataManager() *DataManager {
+	var ret DataManager
+	return &ret
+}
+
+func (dm *DataManager) DownloadRateCapture() *time.Ticker {
+	return dm.rateCapture(dm.GetTotalDownload, &dm.DownloadRateSequence)
+}
+
+func (dm *DataManager) UploadRateCapture() *time.Ticker {
+	return dm.rateCapture(dm.GetTotalUpload, &dm.UploadRateSequence)
+}
+
+func (dm *DataManager) rateCapture(rateFunc func() int64, dst *[]float64) *time.Ticker {
+	ticker := time.NewTicker(rateCaptureFrequency)
+	oldTotal := rateFunc()
+	step := float64(time.Second / rateCaptureFrequency)
+	go func() {
+		for range ticker.C {
+			newTotal := rateFunc()
+			delta := newTotal - oldTotal
+			oldTotal = newTotal
+			rate := float64(delta) * 8 / 1000000 * step // 125000
+			*dst = append(*dst, rate)
+		}
+	}()
+	return ticker
+}
+
+func (dm *DataManager) NewDataChunk() *DataChunk {
+	var dc DataChunk
+	dc.manager = dm
+	dm.Lock()
+	dm.DataGroup = append(dm.DataGroup, &dc)
+	dm.Unlock()
+	return &dc
+}
+
+func (dm *DataManager) GetTotalDownload() int64 {
+	return dm.totalDownload
+}
+
+func (dm *DataManager) GetTotalUpload() int64 {
+	return dm.totalUpload
+}
+
+func (dm *DataManager) Reset() int64 {
+	dm.totalDownload = 0
+	dm.totalUpload = 0
+	dm.DataGroup = []*DataChunk{}
+	dm.DownloadRateSequence = []float64{}
+	dm.UploadRateSequence = []float64{}
+	return dm.totalUpload
+}
+
+func (dm *DataManager) GetAvgDownloadRate() float64 {
+	return calcMAFilter(dm.DownloadRateSequence)
+}
+
+func (dm *DataManager) GetAvgUploadRate() float64 {
+	return calcMAFilter(dm.UploadRateSequence)
+}
+
+type DataChunk struct {
+	manager   *DataManager
+	dateType  DataType
+	startTime time.Time
+	endTime   time.Time
+	dataSize  int64
+	err       error
+
+	ContentLength int64
+	n             int
+}
+
+var blackHolePool = sync.Pool{
+	New: func() any {
+		b := make([]byte, 8192)
+		return &b
+	},
+}
+
+func (dc *DataChunk) GetDuration() time.Duration {
+	return dc.endTime.Sub(dc.startTime)
+}
+
+// DownloadSnapshotHandler No value will be returned here, because the error will interrupt the test.
+// The error chunk is generally caused by the remote server actively closing the connection.
+func (dc *DataChunk) DownloadSnapshotHandler(r io.Reader) error {
+	dc.dateType = TypeDownload
+	dc.startTime = time.Now()
+	defer func() {
+		dc.endTime = time.Now()
+	}()
+	bufP := blackHolePool.Get().(*[]byte)
+	readSize := 0
+	for {
+		readSize, dc.err = r.Read(*bufP)
+		rs := int64(readSize)
+		dc.dataSize += rs
+		atomic.AddInt64(&dc.manager.totalDownload, rs)
+		if dc.err != nil {
+			blackHolePool.Put(bufP)
+			if dc.err == io.EOF {
+				return nil
+			}
+			return dc.err
+		}
+	}
+}
+
+func (dc *DataChunk) UploadSnapshotHandler(size int) *DataChunk {
+	if size <= 0 {
+		panic("the size of repeated bytes should be > 0")
+	}
+
+	dc.ContentLength = int64(size)
+	dc.n = size
+	dc.dateType = TypeUpload
+
+	if dc.manager.repeatByte == nil {
+		r := bytes.Repeat([]byte{0xAA}, readChunkSize) // uniformly distributed sequence of bits
+		dc.manager.repeatByte = &r
+	}
+
+	dc.startTime = time.Now()
+	return dc
+}
+
+func (dc *DataChunk) Read(b []byte) (n int, err error) {
+	if dc.n < readChunkSize {
+		if dc.n <= 0 {
+			dc.endTime = time.Now()
+			return n, io.EOF
+		}
+		n = copy(b, (*dc.manager.repeatByte)[:dc.n])
+	} else {
+		n = copy(b, *dc.manager.repeatByte)
+	}
+	dc.n -= n
+	atomic.AddInt64(&dc.manager.totalUpload, int64(n))
+	return
+}
+
+// calcMAFilter Median-Averaging Filter
+func calcMAFilter(list []float64) float64 {
+	sum := 0.0
+	n := len(list)
+	if n == 0 {
+		return 0
+	}
+	sort.Float64s(list)
+	for i := 1; i < n-1; i++ {
+		sum += list[i]
+	}
+	return sum / float64(n-2)
+}
+
+var GlobalDataManager = NewDataManager()

speedtest/data_manager_test.go

@@ -0,0 +1,60 @@
+package speedtest
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func BenchmarkDataManager_NewDataChunk(b *testing.B) {
+	dmp := NewDataManager()
+	dmp.DataGroup = make([]*DataChunk, 64)
+	for i := 0; i < b.N; i++ {
+		dmp.NewDataChunk()
+
+	}
+}
+
+func TestDynamicRate(t *testing.T) {
+
+	oldDownTotal := GlobalDataManager.GetTotalDownload()
+	oldUpTotal := GlobalDataManager.GetTotalUpload()
+
+	go func() {
+		for i := 0; i < 2; i++ {
+			time.Sleep(time.Second)
+			newDownTotal := GlobalDataManager.GetTotalDownload()
+			newUpTotal := GlobalDataManager.GetTotalUpload()
+
+			downRate := float64(newDownTotal-oldDownTotal) * 8 / 1000 / 1000
+			upRate := float64(newUpTotal-oldUpTotal) * 8 / 1000 / 1000
+			oldDownTotal = newDownTotal
+			oldUpTotal = newUpTotal
+			fmt.Printf("downRate: %.2fMbps | upRate: %.2fMbps\n", downRate, upRate)
+		}
+	}()
+
+	server, _ := CustomServer("http://shenzhen.cmcc.speedtest.shunshiidc.com:8080/speedtest/upload.php")
+	//server, _ := CustomServer("http://192.168.5.237:8080/speedtest/upload.php")
+
+	err := server.DownloadTest(false)
+	if err != nil {
+		fmt.Println("not found server")
+		//t.Error(err)
+	}
+
+	err = server.UploadTest(false)
+	if err != nil {
+		fmt.Println("not found server")
+		//t.Error(err)
+	}
+
+	fmt.Printf(" \n")
+
+	fmt.Printf("Download: %5.2f Mbit/s\n", server.DLSpeed)
+	fmt.Printf("Upload: %5.2f Mbit/s\n\n", server.ULSpeed)
+	valid := server.CheckResultValid()
+	if !valid {
+		fmt.Println("Warning: Result seems to be wrong. Please speedtest again.")
+	}
+}