feat: packet loss analyzer implementation

example/packet_loss/main.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"fmt"
+	"github.com/showwin/speedtest-go/speedtest"
+	"github.com/showwin/speedtest-go/speedtest/transport"
+	"log"
+	"sync"
+)
+
+// Note: The current packet loss analyzer does not support udp over http.
+// This means we cannot get packet loss through a proxy.
+func main() {
+	// 0. Fetching servers
+	serverList, err := speedtest.FetchServers()
+	checkError(err)
+
+	// 1. Retrieve available servers
+	targets := serverList.Available()
+
+	// 2. Create a packet loss analyzer, use default options
+	analyzer, err := speedtest.NewPacketLossAnalyzer(nil)
+	checkError(err)
+
+	wg := &sync.WaitGroup{}
+	// 3. Perform packet loss analysis on all available servers
+	for _, server := range *targets {
+		wg.Add(1)
+		//ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
+		//go func(server *speedtest.Server, analyzer *speedtest.PacketLossAnalyzer, ctx context.Context, cancel context.CancelFunc) {
+		go func(server *speedtest.Server, analyzer *speedtest.PacketLossAnalyzer) {
+			//defer cancel()
+			defer wg.Done()
+			// Note: Please call ctx.cancel at the appropriate time to release resources if you use analyzer.RunWithContext
+			// we using analyzer.Run() here.
+			err = analyzer.Run(server.Host, func(packetLoss *transport.PLoss) {
+				fmt.Println(packetLoss, server.Host, server.Name)
+			})
+			//err = analyzer.RunWithContext(ctx, server.Host, func(packetLoss *transport.PLoss) {
+			//	fmt.Println(packetLoss, server.Host, server.Name)
+			//})
+			if err != nil {
+				fmt.Println(err)
+			}
+			//}(server, analyzer, ctx, cancel)
+		}(server, analyzer)
+	}
+	wg.Wait()
+}
+
+func checkError(err error) {
+	if err != nil {
+		log.Fatal(err)
+	}
+}

speedtest.go

@@ -2,7 +2,9 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"github.com/showwin/speedtest-go/speedtest/transport"
 	"gopkg.in/alecthomas/kingpin.v2"
 	"os"
 	"strconv"
@@ -128,6 +130,24 @@ func main() {
 			task.Printf("Latency: %v Jitter: %v Min: %v Max: %v", server.Latency, server.Jitter, server.MinLatency, server.MaxLatency)
 			task.Complete()
 		})
+
+		// 3.0 create a packet loss analyzer, use default options
+		var analyzer *speedtest.PacketLossAnalyzer
+		analyzer, err = speedtest.NewPacketLossAnalyzer(&speedtest.PacketLossAnalyzerOptions{
+			SourceInterface: *source,
+		})
+		server.PacketLoss = -1.0 // N/A as default
+		packetLossAnalyzerCtx, packetLossAnalyzerCancel := context.WithTimeout(context.Background(), time.Second*40)
+		go func() {
+			err = analyzer.RunWithContext(packetLossAnalyzerCtx, server.Host, func(packetLoss *transport.PLoss) {
+				server.PacketLoss = packetLoss.Loss()
+			})
+			if errors.Is(err, transport.ErrUnsupported) {
+				packetLossAnalyzerCancel() // cancel early
+			}
+		}()
+
+		// 3.1 create accompany Echo
 		accEcho := newAccompanyEcho(server, time.Millisecond*500)
 		taskManager.Run("Download", func(task *Task) {
 			accEcho.Run()
@@ -172,8 +192,15 @@ func main() {
 		})
 		taskManager.Reset()
 		speedtestClient.Manager.Reset()
+		packetLossAnalyzerCancel()
+		if !*jsonOutput {
+			if server.PacketLoss != -1 {
+				fmt.Printf("  Packet Loss: %.2f%%", server.PacketLoss*100)
+			} else {
+				fmt.Printf("  Packet Loss: N/A")
+			}
+		}
 	}
-
 	taskManager.Stop()
 
 	if *jsonOutput {

speedtest/loss.go

@@ -0,0 +1,123 @@
+package speedtest
+
+import (
+	"context"
+	"github.com/showwin/speedtest-go/speedtest/transport"
+	"net"
+	"time"
+)
+
+type PacketLossAnalyzerOptions struct {
+	RemoteSamplingInterval time.Duration
+	SamplingDuration       time.Duration
+	PacketSendingInterval  time.Duration
+	PacketSendingTimeout   time.Duration
+	SourceInterface        string      // source interface
+	TCPDialer              *net.Dialer // tcp dialer for sampling
+	UDPDialer              *net.Dialer // udp dialer for sending packet
+
+}
+
+type PacketLossAnalyzer struct {
+	options *PacketLossAnalyzerOptions
+}
+
+func NewPacketLossAnalyzer(options *PacketLossAnalyzerOptions) (*PacketLossAnalyzer, error) {
+	if options == nil {
+		options = &PacketLossAnalyzerOptions{}
+	}
+	if options.SamplingDuration == 0 {
+		options.SamplingDuration = time.Second * 30
+	}
+	if options.RemoteSamplingInterval == 0 {
+		options.RemoteSamplingInterval = 1 * time.Second
+	}
+	if options.PacketSendingInterval == 0 {
+		options.PacketSendingInterval = 67 * time.Millisecond
+	}
+	if options.PacketSendingTimeout == 0 {
+		options.PacketSendingTimeout = 5 * time.Second
+	}
+	if options.TCPDialer == nil {
+		options.TCPDialer = &net.Dialer{
+			Timeout: options.PacketSendingTimeout,
+		}
+	}
+	if options.UDPDialer == nil {
+		var addr net.Addr
+		if len(options.SourceInterface) > 0 {
+			// skip error and using auto-select
+			addr, _ = net.ResolveUDPAddr("udp", options.SourceInterface)
+		}
+		options.UDPDialer = &net.Dialer{
+			Timeout:   options.PacketSendingTimeout,
+			LocalAddr: addr,
+		}
+	}
+	return &PacketLossAnalyzer{
+		options: options,
+	}, nil
+}
+
+func (pla *PacketLossAnalyzer) Run(host string, callback func(packetLoss *transport.PLoss)) error {
+	ctx, cancel := context.WithTimeout(context.Background(), pla.options.SamplingDuration)
+	defer cancel()
+	return pla.RunWithContext(ctx, host, callback)
+}
+
+func (pla *PacketLossAnalyzer) RunWithContext(ctx context.Context, host string, callback func(packetLoss *transport.PLoss)) error {
+	samplerClient, err := transport.NewClient(pla.options.TCPDialer)
+	if err != nil {
+		return transport.ErrUnsupported
+	}
+	senderClient, err := transport.NewPacketLossSender(samplerClient.ID(), pla.options.UDPDialer)
+	if err != nil {
+		return transport.ErrUnsupported
+	}
+
+	if err = samplerClient.Connect(ctx, host); err != nil {
+		return transport.ErrUnsupported
+	}
+	if err = senderClient.Connect(ctx, host); err != nil {
+		return transport.ErrUnsupported
+	}
+	if err = samplerClient.InitPacketLoss(); err != nil {
+		return transport.ErrUnsupported
+	}
+	go pla.loopSender(ctx, senderClient)
+	return pla.loopSampler(ctx, samplerClient, callback)
+}
+
+func (pla *PacketLossAnalyzer) loopSampler(ctx context.Context, client *transport.Client, callback func(packetLoss *transport.PLoss)) error {
+	ticker := time.NewTicker(pla.options.RemoteSamplingInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if pl, err1 := client.PacketLoss(); err1 == nil {
+				if pl != nil {
+					callback(pl)
+				}
+			} else {
+				return err1
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+func (pla *PacketLossAnalyzer) loopSender(ctx context.Context, senderClient *transport.PacketLossSender) {
+	order := 0
+	sendTick := time.NewTicker(pla.options.PacketSendingInterval)
+	defer sendTick.Stop()
+	for {
+		select {
+		case <-sendTick.C:
+			_ = senderClient.Send(order)
+			order++
+		case <-ctx.Done():
+			return
+		}
+	}
+}

speedtest/request.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/showwin/speedtest-go/speedtest/transport"
 	"io"
 	"math"
 	"net/http"
@@ -12,8 +13,6 @@ import (
 	"strings"
 	"sync/atomic"
 	"time"
-
-	"github.com/showwin/speedtest-go/speedtest/tcp"
 )
 
 type (
@@ -280,8 +279,11 @@ func (s *Server) TCPPing(
 		pingDst = s.Host
 	}
 	failTimes := 0
-	client := tcp.NewClient(s.Context.tcpDialer, pingDst)
-	err = client.Connect()
+	client, err := transport.NewClient(s.Context.tcpDialer)
+	if err != nil {
+		return nil, err
+	}
+	err = client.Connect(ctx, pingDst)
 	if err != nil {
 		return nil, err
 	}

speedtest/server.go

@@ -51,6 +51,7 @@ type Server struct {
 	DLSpeed      ByteRate      `json:"dl_speed"`
 	ULSpeed      ByteRate      `json:"ul_speed"`
 	TestDuration TestDuration  `json:"test_duration"`
+	PacketLoss   float64       `json:"packet_loss"`
 
 	Context *Speedtest `json:"-"`
 }