bwtester: convert to pan

Add default path policy options to command line for bwtestclient.

For the data channel, we want to ensure to use the path used for the
control channel and stick to this, i.e. no more fail-over once
connection is established. On the client side, we use the Pin policy to,
well, pin the path for the data channel. On the server side, we
initialize a default reply path selector with the path known from the
control channel. With this approach, we stick to the original forwarding
path, but the path can still be refreshed if it expires (however
unlikely that is, given the short max duration of tests). We could
perhaps go a bit further and use dumb selectors that always return the
fixed, initial path.

Author: matzf

bwtester/bwtestclient/bwtestclient.go

@@ -33,10 +33,10 @@ import (
 	"time"
 	"unicode"
 
+	"inet.af/netaddr"
+
 	. "github.com/netsec-ethz/scion-apps/bwtester/bwtestlib"
-	"github.com/netsec-ethz/scion-apps/pkg/appnet"
-	"github.com/scionproto/scion/go/lib/addr"
-	"github.com/scionproto/scion/go/lib/snet"
+	"github.com/netsec-ethz/scion-apps/pkg/pan"
 )
 
 const (
@@ -128,7 +128,7 @@ func parseBwtestParameters(s string, defaultPktSize int64) (BwtestParameters, er
 			a4 = parseBandwidth(a[3])
 			a2 = (a4 * a1) / (a3 * 8)
 		} else {
-			a2 = int64(defaultPktSize)
+			a2 = defaultPktSize
 		}
 	} else {
 		a2 = getPacketSize(a[1], defaultPktSize)
@@ -248,83 +248,83 @@ func getPacketCount(count string) int64 {
 	return a3
 }
 
+func usageErr(msg string) {
+	printUsage()
+	if msg != "" {
+		fmt.Println("\nError:", msg)
+	}
+	os.Exit(2)
+}
+
+func checkUsageErr(err error) {
+	if err != nil {
+		usageErr(err.Error())
+	}
+}
+
 func main() {
 	var (
-		serverCCAddrStr string
-		clientBwpStr    string
-		serverBwpStr    string
-		interactive     bool
-		pathAlgo        string
+		local        pan.IPPortValue
+		serverCCAddr pan.UDPAddr
+		clientBwpStr string
+		serverBwpStr string
+		interactive  bool
+		sequence     string
+		preference   string
 	)
 
 	flag.Usage = printUsage
-	flag.StringVar(&serverCCAddrStr, "s", "", "Server SCION Address")
+	flag.Var(&local, "local", "Local address")
+	flag.Var(&serverCCAddr, "s", "Server SCION Address")
 	flag.StringVar(&serverBwpStr, "sc", DefaultBwtestParameters, "Server->Client test parameter")
 	flag.StringVar(&clientBwpStr, "cs", DefaultBwtestParameters, "Client->Server test parameter")
 	flag.BoolVar(&interactive, "i", false, "Interactive path selection, prompt to choose path")
-	flag.StringVar(&pathAlgo, "pathAlgo", "", "Path selection algorithm / metric (\"shortest\", \"mtu\")")
+	flag.StringVar(&sequence, "sequence", "", "Sequence of space separated hop predicates to specify path")
+	flag.StringVar(&preference, "preference", "", "Preference sorting order for paths. "+
+		"Comma-separated list of available sorting options: "+
+		strings.Join(pan.AvailablePreferencePolicies, "|"))
 
 	flag.Parse()
 	flagset := make(map[string]bool)
 	// record if flags were set or if default value was used
 	flag.Visit(func(f *flag.Flag) { flagset[f.Name] = true })
 
 	if flag.NFlag() == 0 {
-		// no flag was set, only print usage and exit
-		printUsage()
-		os.Exit(2)
+		usageErr("")
 	}
-	if len(serverCCAddrStr) == 0 {
-		printUsage()
-		fmt.Println("\nServer address needs to be specified with -s")
-		os.Exit(2)
+	if !serverCCAddr.IsValid() {
+		usageErr("server address needs to be specified with -s")
 	}
-	serverCCAddr, err := appnet.ResolveUDPAddr(serverCCAddrStr)
-	Check(err)
+	policy, err := pan.PolicyFromCommandline(sequence, preference, interactive)
+	checkUsageErr(err)
 
-	var path snet.Path
-	if interactive {
-		path, err = appnet.ChoosePathInteractive(serverCCAddr.IA)
-		Check(err)
-	} else {
-		var metric int
-		if pathAlgo == "mtu" {
-			metric = appnet.MTU
-		} else if pathAlgo == "shortest" {
-			metric = appnet.Shortest
-		}
-		path, err = appnet.ChoosePathByMetric(metric, serverCCAddr.IA)
-		Check(err)
-	}
-	if path != nil {
-		appnet.SetPath(serverCCAddr, path)
-	}
-
-	// use default packet size when within same AS and pathEntry is not set
+	// use default packet size when within same AS
 	inferedPktSize := int64(DefaultPktSize)
 	// update default packet size to max MTU on the selected path
-	if path != nil {
+	// TODO(matzf): evaluate policy, set pkt size to MTU of most preferred path,
+	//              append filter to policy to allow only paths with MTU >= pkt size.
+	/*if path != nil {
 		inferedPktSize = int64(path.Metadata().MTU)
-	}
+	}*/
 	if !flagset["cs"] && flagset["sc"] { // Only one direction set, used same for reverse
 		clientBwpStr = serverBwpStr
 		fmt.Println("Only sc parameter set, using same values for cs")
 	}
 	clientBwp, err := parseBwtestParameters(clientBwpStr, inferedPktSize)
-	Check(err)
+	checkUsageErr(err)
 	if !flagset["sc"] && flagset["cs"] { // Only one direction set, used same for reverse
 		serverBwpStr = clientBwpStr
 		fmt.Println("Only cs parameter set, using same values for sc")
 	}
 	serverBwp, err := parseBwtestParameters(serverBwpStr, inferedPktSize)
-	Check(err)
+	checkUsageErr(err)
 	fmt.Println("\nTest parameters:")
 	fmt.Printf("client->server: %d seconds, %d bytes, %d packets\n",
 		int(clientBwp.BwtestDuration/time.Second), clientBwp.PacketSize, clientBwp.NumPackets)
 	fmt.Printf("server->client: %d seconds, %d bytes, %d packets\n",
 		int(serverBwp.BwtestDuration/time.Second), serverBwp.PacketSize, serverBwp.NumPackets)
 
-	clientRes, serverRes, err := runBwtest(serverCCAddr, clientBwp, serverBwp)
+	clientRes, serverRes, err := runBwtest(local.Get(), serverCCAddr, policy, clientBwp, serverBwp)
 	Check(err)
 
 	fmt.Println("\nS->C results")
@@ -334,31 +334,29 @@ func main() {
 }
 
 // runBwtest runs the bandwidth test with the given parameters against the server at serverCCAddr.
-func runBwtest(serverCCAddr *snet.UDPAddr,
+func runBwtest(local netaddr.IPPort, serverCCAddr pan.UDPAddr, policy pan.Policy,
 	clientBwp, serverBwp BwtestParameters) (clientRes, serverRes BwtestResult, err error) {
 
 	// Control channel connection
-	ccConn, err := appnet.DialAddr(serverCCAddr)
+	ccSelector := pan.NewDefaultSelector()
+	ccConn, err := pan.DialUDP(context.Background(), local, serverCCAddr, policy, ccSelector)
 	if err != nil {
 		return
 	}
-	clientCCAddr := ccConn.LocalAddr().(*net.UDPAddr)
 
-	// Address of client data channel (DC)
-	clientDCAddr := &net.UDPAddr{IP: clientCCAddr.IP, Port: clientCCAddr.Port + 1}
+	dcLocal := local.WithPort(0)
 	// Address of server data channel (DC)
-	serverDCAddr := serverCCAddr.Copy()
-	serverDCAddr.Host.Port = serverCCAddr.Host.Port + 1
-	// DC ports are passed in the request
-	clientBwp.Port = uint16(clientDCAddr.Port)
-	serverBwp.Port = uint16(serverDCAddr.Host.Port)
+	serverDCAddr := serverCCAddr.WithPort(serverCCAddr.Port + 1)
 
 	// Data channel connection
-	dcConn, err := appnet.DefNetwork().Dial(
-		context.TODO(), "udp", clientDCAddr, serverDCAddr, addr.SvcNone)
+	dcConn, err := pan.DialUDP(context.Background(), dcLocal, serverDCAddr, policy, nil)
 	if err != nil {
 		return
 	}
+	clientDCAddr := dcConn.LocalAddr().(pan.UDPAddr)
+	// DC ports are passed in the request
+	clientBwp.Port = clientDCAddr.Port
+	serverBwp.Port = serverDCAddr.Port
 
 	// Start receiver before even sending the request so it will be ready.
 	receiveRes := make(chan BwtestResult, 1)
@@ -383,6 +381,11 @@ func runBwtest(serverCCAddr *snet.UDPAddr,
 		return
 	}
 
+	// Pin DC to path used for request
+	if serverDCAddr.IA != clientDCAddr.IA {
+		dcConn.SetPolicy(pan.Pinned{ccSelector.Path().Fingerprint})
+	}
+
 	// Start blasting client->server
 	err = HandleDCConnSend(clientBwp, dcConn)
 	if err != nil && !errors.Is(err, os.ErrDeadlineExceeded) {

bwtester/bwtestclient/bwtestclient_integration_test.go

@@ -36,10 +36,9 @@ func TestIntegrationBwtestclient(t *testing.T) {
 	serverCmd := integration.AppBinPath(serverBin)
 
 	// Server
-	serverPort := "40002"
-	serverArgs := []string{"-p", serverPort}
+	serverArgs := []string{}
 	// Client
-	clientArgs := []string{"-s", integration.DstAddrPattern + ":" + serverPort, "-cs", "1,?,?,1Mbps"}
+	clientArgs := []string{"-s", integration.DstAddrPattern + ":40002", "-cs", "1,?,?,1Mbps"}
 
 	in := integration.NewAppsIntegration(clientCmd, serverCmd, clientArgs, serverArgs)
 	in.ServerOutMatch = integration.Contains("Received request")

bwtester/bwtestserver/bwtestserver.go

@@ -22,31 +22,31 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"flag"
 	"fmt"
 	"net"
 	"time"
 
+	"gopkg.in/alecthomas/kingpin.v2"
+	"inet.af/netaddr"
+
 	. "github.com/netsec-ethz/scion-apps/bwtester/bwtestlib"
-	"github.com/netsec-ethz/scion-apps/pkg/appnet"
-	"github.com/scionproto/scion/go/lib/addr"
-	"github.com/scionproto/scion/go/lib/snet"
+	"github.com/netsec-ethz/scion-apps/pkg/pan"
 )
 
 const (
 	resultExpiry = time.Minute
 )
 
 func main() {
-	// Fetch arguments from command line
-	serverPort := flag.Uint("p", 40002, "Port")
-	flag.Parse()
+	var listen pan.IPPortValue
+	kingpin.Flag("listen", "Address to listen on").Default(":40002").SetValue(&listen)
+	kingpin.Parse()
 
-	err := runServer(uint16(*serverPort))
+	err := runServer(listen.Get())
 	Check(err)
 }
 
-func runServer(port uint16) error {
+func runServer(listen netaddr.IPPort) error {
 	receivePacketBuffer := make([]byte, 2500)
 
 	var currentBwtest string
@@ -55,11 +55,12 @@ func runServer(port uint16) error {
 
 	results := make(resultsMap)
 
-	ccConn, err := appnet.ListenPort(port)
+	ccSelector := pan.NewDefaultReplySelector()
+	ccConn, err := pan.ListenUDP(context.Background(), listen, ccSelector)
 	if err != nil {
 		return err
 	}
-	serverCCAddr := ccConn.LocalAddr().(*net.UDPAddr)
+	serverCCAddr := ccConn.LocalAddr().(pan.UDPAddr)
 	for {
 		// Handle client requests
 		n, clientCCAddr, err := ccConn.ReadFrom(receivePacketBuffer)
@@ -104,7 +105,8 @@ func runServer(port uint16) error {
 			if err != nil {
 				continue
 			}
-			finishTime, err := startBwtestBackground(serverCCAddr, clientCCAddr.(*snet.UDPAddr),
+			path := ccSelector.Path(clientCCAddr.(pan.UDPAddr))
+			finishTime, err := startBwtestBackground(serverCCAddr, clientCCAddr.(pan.UDPAddr), path,
 				clientBwp, serverBwp, currentResult)
 			if err != nil {
 				// Ask the client to try again in 1 second
@@ -135,17 +137,17 @@ func runServer(port uint16) error {
 
 // startBwtestBackground starts a bandwidth test, in the background.
 // Returns the expected finish time of the test, or any error during the setup.
-func startBwtestBackground(serverCCAddr *net.UDPAddr, clientCCAddr *snet.UDPAddr,
-	clientBwp, serverBwp BwtestParameters, res chan<- BwtestResult) (time.Time, error) {
+func startBwtestBackground(serverCCAddr pan.UDPAddr, clientCCAddr pan.UDPAddr,
+	path *pan.Path, clientBwp, serverBwp BwtestParameters, res chan<- BwtestResult) (time.Time, error) {
 
 	// Data Connection addresses:
-	clientDCAddr := clientCCAddr.Copy()
-	clientDCAddr.Host.Port = int(clientBwp.Port)
-	serverDCAddr := &net.UDPAddr{IP: serverCCAddr.IP, Port: int(serverBwp.Port)}
+	clientDCAddr := clientCCAddr
+	clientDCAddr.Port = clientBwp.Port
+	serverDCAddr := netaddr.IPPortFrom(serverCCAddr.IP, serverBwp.Port)
 
 	// Open Data Connection
-	dcConn, err := appnet.DefNetwork().Dial(
-		context.TODO(), "udp", serverDCAddr, clientDCAddr, addr.SvcNone)
+	dcSelector := initializedReplySelector(clientDCAddr, path)
+	dcConn, err := listenConnected(serverDCAddr, clientDCAddr, dcSelector)
 	if err != nil {
 		return time.Time{}, err
 	}
@@ -287,3 +289,50 @@ func (r resultsMap) purgeExpired() {
 		}
 	}
 }
+
+func listenConnected(local netaddr.IPPort, remote pan.UDPAddr, selector pan.ReplySelector) (net.Conn, error) {
+	conn, err := pan.ListenUDP(context.Background(), local, selector)
+	return connectedPacketConn{
+		ListenConn: conn,
+		remote:     remote,
+	}, err
+}
+
+// connectedPacketConn connects a net.PacketConn to a fixed remote address,
+// i.e. it uses the fixed remote address to wrap the ReadFrom/WriteTo of a
+// net.PacketConn into Read/Write of a net.Conn.
+type connectedPacketConn struct {
+	pan.ListenConn
+	remote pan.UDPAddr
+}
+
+func (c connectedPacketConn) Read(buf []byte) (int, error) {
+	for {
+		n, addr, err := c.ListenConn.ReadFrom(buf)
+		if err != nil {
+			return n, err
+		}
+		if c.remote != addr.(pan.UDPAddr) {
+			continue
+		}
+		return n, err
+	}
+}
+
+func (c connectedPacketConn) Write(buf []byte) (int, error) {
+	return c.ListenConn.WriteTo(buf, c.remote)
+}
+
+func (c connectedPacketConn) RemoteAddr() net.Addr {
+	return c.remote
+}
+
+// initializedReplySelector creates a pan.DefaultReplySelector, initialized with path for dst.
+func initializedReplySelector(remote pan.UDPAddr, path *pan.Path) pan.ReplySelector {
+	if path != nil && path.Destination != remote.IA {
+		panic("path destination should match address")
+	}
+	selector := pan.NewDefaultReplySelector()
+	selector.Record(remote, path)
+	return selector
+}