Add support for throttling in Bandwidth tests, ToS and improved external mode. (#135)

* Intermediate checkin

* Last check-in before deleting ACK message in Ethr.

* Support for client port, throttling and tos almost working.

* Most functionality working as expected with code all cleaned up.

* Linux/OSX Fixes.

* Fix handshake mechanism.

* Minor cleanup.

* More improvements for external mode.

* Improve admin-mode, root user permission checking.

* Improve detection of IP version for ICMP.

* Update README.md

README.md

@@ -138,18 +138,25 @@ ethr -c localhost -n 8
 // Start connections/s test using 64 threads to server 10.1.0.11
 ethr -c 10.1.0.11 -t c -n 64
 
+// Run Ethr server on port 9999
+./ethr -s -port 9999
+
+// Measure TCP connection setup latency to ethr server on port 9999
+// Assuming Ethr server is running on server with IP address: 10.1.1.100
+./ethr -c 10.1.1.100 -p tcp -t pi -d 0 -4 -port 9999
+
 // Measure TCP connection setup latency to www.github.com at port 443
-./ethr -c www.github.com:443 -p tcp -t pi -m x -d 0 -4
+./ethr -x www.github.com:443 -p tcp -t pi -d 0 -4
 
 // Measure TCP connection setup latency to www.github.com at port 443
 // Note: Here port 443 is driven automatically from https
-./ethr -c https://www.github.com -p tcp -t pi -m x -d 0 -4
+./ethr -x https://www.github.com -p tcp -t pi -d 0 -4
 
 // Measure ICMP ping latency to www.github.com
-sudo ./ethr -c www.github.com -p icmp -t pi -m x -d 0 -4
+sudo ./ethr -x www.github.com -p icmp -t pi -d 0 -4
 
 // Run measurement similar to mtr on Linux
-sudo ./ethr -c www.github.com -p icmp -t mtr -m x -d 0 -4
+sudo ./ethr -x www.github.com -p icmp -t mtr -d 0 -4
 
 // Measure packets/s over UDP by sending small 1-byte packets
 ./ethr -c 172.28.192.1 -p udp -t p -d 0
@@ -187,24 +194,35 @@ For ICMP Ping, ICMP/TCP TraceRoute and MyTraceRoute, privileged mode is required
 	-6 
 		Use only IP v6 version
 ```
-### Server Parameters
+### Server Mode Parameters
 ```
 In this mode, Ethr runs as a server, allowing multiple clients to run
 performance tests against it.
 	-s 
 		Run in server mode.
-	-ui 
+	-ip <string>
-		Show output in text UI.
+		Bind to specified local IP address for TCP & UDP tests.
+		This must be a valid IPv4 or IPv6 address.
+		Default: <empty> - Any IP
 	-port <number>
 		Use specified port number for TCP & UDP tests.
 		Default: 8888
+	-ui 
+		Show output in text UI.
 ```
-### Client Parameters
+### Client Mode Parameters
 ```
 In this mode, Ethr client can only talk to an Ethr server.
 	-c <server>
 		Run in client mode and connect to <server>.
 		Server is specified using name, FQDN or IP address.
+	-b <rate>
+		Bytes to send per second (format: <num>[KB | MB | GB])
+		Only valid for Bandwidth tests. Default: 0 - Unlimited
+		Examples: 100 (100B/s or 800bits/s), 1MB (1MB/s or 8Mbits/s).
+	-cport <number>
+		Use specified local port number in client for TCP & UDP tests.
+		Default: 0 - Ephemeral Port
 	-d <duration>
 		Duration for the test (format: <num>[ms | s | m | h]
 		0: Run forever
@@ -218,6 +236,10 @@ In this mode, Ethr client can only talk to an Ethr server.
 		Number of round trip iterations for each latency measurement.
 		Only valid for latency testing.
 		Default: 1000
+	-ip <string>
+		Bind to specified local IP address for TCP & UDP tests.
+		This must be a valid IPv4 or IPv6 address.
+		Default: <empty> - Any IP
 	-l <length>
 		Length of buffer to use (format: <num>[KB | MB | GB])
 		Only valid for Bandwidth tests. Max 1GB.
@@ -242,23 +264,27 @@ In this mode, Ethr client can only talk to an Ethr server.
 		l: Latency, Loss & Jitter
 		pi: Ping Loss & Latency
 		tr: TraceRoute
-                mtr: MyTraceRoute with Loss & Latency
+		mtr: MyTraceRoute with Loss & Latency
 		Default: b - Bandwidth measurement.
+	-tos 
+		Specifies 8-bit value to use in IPv4 TOS field or IPv6 Traffic Class field.
 	-w <number>
 		Use specified number of iterations for warmup.
 		Default: 1
 ```
-### External Client Mode
+### External Mode Parameters
 ```
-In this mode, Ethr client can talk to a non-Ethr server. This mode only supports
+In this mode, Ethr talks to a non-Ethr server. This mode supports only a
 few types of measurements, such as Ping, Connections/s and TraceRoute.
-	-c <destination>
+	-x <destination>
 		Run in external client mode and connect to <destination>.
-		<destination> is specified in <host:port> format for TCP and <host> format for ICMP.
+		<destination> is specified in URL or Host:Port format.
-		Example: For TCP - www.microsoft.com:443 or 10.1.0.4:22
+		For URL, if port is not specified, it is assumed to be 80 for http and 443 for https.
+		Example: For TCP - www.microsoft.com:443 or 10.1.0.4:22 or https://www.github.com
 		         For ICMP - www.microsoft.com or 10.1.0.4
-	-m <mode>
+	-cport <number>
-		'-m x' MUST be specified for external mode.
+		Use specified local port number in client for TCP & UDP tests.
+		Default: 0 - Ephemeral Port
 	-d <duration>
 		Duration for the test (format: <num>[ms | s | m | h]
 		0: Run forever
@@ -268,6 +294,10 @@ few types of measurements, such as Ping, Connections/s and TraceRoute.
 		Only valid for latency, ping and traceRoute tests.
 		0: No gap
 		Default: 1s
+	-ip <string>
+		Bind to specified local IP address for TCP & UDP tests.
+		This must be a valid IPv4 or IPv6 address.
+		Default: <empty> - Any IP
 	-n <number>
 		Number of Parallel Sessions (and Threads).
 		0: Equal to number of CPUs
@@ -280,8 +310,10 @@ few types of measurements, such as Ping, Connections/s and TraceRoute.
 		c: Connections/s
 		pi: Ping Loss & Latency
 		tr: TraceRoute
-                mtr: MyTraceRoute with Loss & Latency
+		mtr: MyTraceRoute with Loss & Latency
 		Default: pi - Ping Loss & Latency.
+	-tos 
+		Specifies 8-bit value to use in IPv4 TOS field or IPv6 Traffic Class field.
 	-w <number>
 		Use specified number of iterations for warmup.
 		Default: 1
@@ -293,7 +325,7 @@ Protocol  | Bandwidth | Connections/s | Packets/s | Latency | Ping | TraceRoute
 ------------- | ------------- | ------------- | ------------- | ------------- | ------------- | ------------- | -------------
 TCP  | Yes | Yes | NA | Yes | Yes | Yes | Yes
 UDP  | Yes | NA | Yes | No | NA | No | No
-ICMP | No | NA | No | No | Yes | Yes | Yes
+ICMP | No | NA | NA | NA | Yes | Yes | Yes
 
 # Platform Support
 
@@ -322,10 +354,7 @@ No other platforms are tested at this time
 Todo list work items are shown below. Contributions are most welcome for these work items or any other features and bugfixes.
 
 * Test Ethr on other Windows versions, other Linux versions, FreeBSD and other OS
-* Support for UDP bandwidth & latency testing
+* Support for UDP latency, TraceRoute and MyTraceRoute
-* Support for HTTPS bandwidth, latency, requests/s
-* Support for HTTP latency and requests/s
-* Support for ICMP bandwidth, latency and packets/s
 
 # Contributing
 

client.go

@@ -12,7 +12,6 @@ import (
 
 	"bytes"
 	"encoding/binary"
-	"encoding/gob"
 	"encoding/hex"
 	"fmt"
 	"io"
@@ -74,26 +73,22 @@ func initClient() {
 	initClientUI()
 }
 
-func handshakeWithServer(test *ethrTest, conn net.Conn) {
+func handshakeWithServer(test *ethrTest, conn net.Conn) (err error) {
-	dec := gob.NewDecoder(conn)
+	ethrMsg := createSynMsg(test.testID, test.clientParam)
-	enc := gob.NewEncoder(conn)
+	err = sendSessionMsg(conn, ethrMsg)
-	ethrMsg := createSynMsg(test.testParam)
-	err := sendSessionMsg(enc, ethrMsg)
 	if err != nil {
-		ui.printErr("Failed to send session message: %v", err)
+		ui.printDbg("Failed to send SYN message to Ethr server. Error: %v", err)
 		return
 	}
-	ethrMsg = recvSessionMsg(dec)
+	ethrMsg = recvSessionMsg(conn)
 	if ethrMsg.Type != EthrAck {
-		if ethrMsg.Type == EthrFin {
+		ui.printDbg("Failed to receive ACK message from Ethr server. Error: %v", err)
-			err = fmt.Errorf("%s", ethrMsg.Fin.Message)
+		err = os.ErrInvalid
-		} else {
-			err = fmt.Errorf("Unexpected control message received. %v", ethrMsg)
-		}
 	}
+	return
 }
 
-func getServerIPandPort(server string) (string, string, error) {
+func getServerIPandPort(server string) (string, string, string, error) {
 	hostName := ""
 	hostIP := ""
 	port := ""
@@ -103,10 +98,13 @@ func getServerIPandPort(server string) (string, string, error) {
 		if u.Port() != "" {
 			port = u.Port()
 		} else {
-			if u.Scheme == "http" {
+			// Only implicitly derive port in External client mode.
-				port = "80"
+			if gIsExternalClient {
-			} else if u.Scheme == "https" {
+				if u.Scheme == "http" {
-				port = "443"
+					port = "80"
+				} else if u.Scheme == "https" {
+					port = "443"
+				}
 			}
 		}
 	} else {
@@ -116,21 +114,41 @@ func getServerIPandPort(server string) (string, string, error) {
 		}
 	}
 	_, hostIP, err = ethrLookupIP(hostName)
-	return hostIP, port, err
+	return hostName, hostIP, port, err
 }
 
-func runClient(testParam EthrTestParam, clientParam ethrClientParam, server string) {
+func runClient(testID EthrTestID, clientParam EthrClientParam, server string) {
 	initClient()
-	hostIP, port, err := getServerIPandPort(server)
+	hostName, hostIP, port, err := getServerIPandPort(server)
 	if err != nil {
 		return
 	}
-	if !xMode {
+	ip := net.ParseIP(hostIP)
-		// For Ethr to Ethr tests, override the port supplied as part
+	if ip != nil {
-		// of the server name/url.
+		if ip.To4() != nil {
+			gIPVersion = ethrIPv4
+		} else {
+			gIPVersion = ethrIPv6
+		}
+	} else {
+		return
+	}
+
+	if gIsExternalClient {
+		if testID.Protocol != ICMP && port == "" {
+			ui.printErr("In external mode, port cannot be empty for TCP tests.")
+			return
+		}
+	} else {
+		if port != "" {
+			ui.printErr("In client mode, port (%s) cannot be specified in destination (%s).", port, server)
+			ui.printMsg("Hint: Use external mode (-x).")
+			return
+		}
 		port = gEthrPortStr
 	}
-	test, err := newTest(hostIP, nil, testParam)
+	ui.printMsg("Using destination: %s, ip: %s, port: %s", hostName, hostIP, port)
+	test, err := newTest(hostIP, testID, clientParam)
 	if err != nil {
 		ui.printErr("Failed to create the new test.")
 		return
@@ -138,60 +156,65 @@ func runClient(testParam EthrTestParam, clientParam ethrClientParam, server stri
 	test.remoteAddr = server
 	test.remoteIP = hostIP
 	test.remotePort = port
-	if testParam.TestID.Protocol == ICMP {
+	if testID.Protocol == ICMP {
 		test.dialAddr = hostIP
 	} else {
 		test.dialAddr = fmt.Sprintf("[%s]:%s", hostIP, port)
 	}
-	runTest(test, clientParam.duration, clientParam.gap, clientParam.warmupCount)
+	runTest(test)
 }
 
-func runTest(test *ethrTest, d, g time.Duration, warmupCount int) {
+func runTest(test *ethrTest) {
-	toStop := make(chan int, 1)
+	toStop := make(chan int, 16)
 	startStatsTimer()
-	runDurationTimer(d, toStop)
+	gap := test.clientParam.Gap
+	duration := test.clientParam.Duration
+	runDurationTimer(duration, toStop)
 	test.isActive = true
-	if test.testParam.TestID.Protocol == TCP {
+	if test.testID.Protocol == TCP {
-		if test.testParam.TestID.Type == Bandwidth {
+		if test.testID.Type == Bandwidth {
 			tcpRunBandwidthTest(test, toStop)
-		} else if test.testParam.TestID.Type == Latency {
+		} else if test.testID.Type == Latency {
-			go runTCPLatencyTest(test, g, toStop)
+			go runTCPLatencyTest(test, gap, toStop)
-		} else if test.testParam.TestID.Type == Cps {
+		} else if test.testID.Type == Cps {
 			go tcpRunCpsTest(test)
-		} else if test.testParam.TestID.Type == Ping {
+		} else if test.testID.Type == Ping {
-			go clientRunPingTest(test, g, warmupCount)
+			go clientRunPingTest(test, gap, test.clientParam.WarmupCount)
-		} else if test.testParam.TestID.Type == TraceRoute {
+		} else if test.testID.Type == TraceRoute {
-			tcpRunTraceRoute(test, g, toStop)
+			VerifyPermissionForTest(test.testID)
-		} else if test.testParam.TestID.Type == MyTraceRoute {
+			go tcpRunTraceRoute(test, gap, toStop)
-			tcpRunMyTraceRoute(test, g, toStop)
+		} else if test.testID.Type == MyTraceRoute {
+			VerifyPermissionForTest(test.testID)
+			go tcpRunMyTraceRoute(test, gap, toStop)
 		}
-	} else if test.testParam.TestID.Protocol == UDP {
+	} else if test.testID.Protocol == UDP {
-		if test.testParam.TestID.Type == Bandwidth ||
+		if test.testID.Type == Bandwidth ||
-			test.testParam.TestID.Type == Pps {
+			test.testID.Type == Pps {
 			runUDPBandwidthAndPpsTest(test)
 		}
-	} else if test.testParam.TestID.Protocol == ICMP {
+	} else if test.testID.Protocol == ICMP {
-		if test.testParam.TestID.Type == Ping {
+		VerifyPermissionForTest(test.testID)
-			go clientRunPingTest(test, g, warmupCount)
+		if test.testID.Type == Ping {
-		} else if test.testParam.TestID.Type == TraceRoute {
+			go clientRunPingTest(test, gap, test.clientParam.WarmupCount)
-			icmpRunTraceRoute(test, g, toStop)
+		} else if test.testID.Type == TraceRoute {
-		} else if test.testParam.TestID.Type == MyTraceRoute {
+			go icmpRunTraceRoute(test, gap, toStop)
-			icmpRunMyTraceRoute(test, g, toStop)
+		} else if test.testID.Type == MyTraceRoute {
+			go icmpRunMyTraceRoute(test, gap, toStop)
 		}
 	}
-
 	handleInterrupt(toStop)
 	reason := <-toStop
 	stopStatsTimer()
 	close(test.done)
-	if test.testParam.TestID.Type == Ping {
+	if test.testID.Type == Ping {
 		time.Sleep(2 * time.Second)
 	}
 	switch reason {
 	case done:
 		ui.printMsg("Ethr done, measurement complete.")
 	case timeout:
-		ui.printMsg("Ethr done, duration: " + d.String() + ".")
+		ui.printMsg("Ethr done, duration: " + duration.String() + ".")
+		ui.printMsg("Hint: Use -d parameter to change duration of the test.")
 	case interrupt:
 		ui.printMsg("Ethr done, received interrupt signal.")
 	case disconnect:
@@ -210,13 +233,18 @@ func tcpRunBandwidthTest(test *ethrTest, toStop chan int) {
 }
 
 func tcpRunBanwidthTestThreads(test *ethrTest, wg *sync.WaitGroup) {
-	for th := uint32(0); th < test.testParam.NumThreads; th++ {
+	for th := uint32(0); th < test.clientParam.NumThreads; th++ {
-		conn, err := net.Dial(tcp(ipVer), test.dialAddr)
+		conn, err := ethrDialInc(TCP, test.dialAddr, uint16(th))
 		if err != nil {
 			ui.printErr("Error dialing connection: %v", err)
-			return
+			continue
+		}
+		err = handshakeWithServer(test, conn)
+		if err != nil {
+			ui.printErr("Failed in handshake with the server. Error: %v", err)
+			conn.Close()
+			continue
 		}
-		handshakeWithServer(test, conn)
 		wg.Add(1)
 		go runTCPBandwidthTestHandler(test, conn, wg)
 	}
@@ -230,11 +258,15 @@ func runTCPBandwidthTestHandler(test *ethrTest, conn net.Conn, wg *sync.WaitGrou
 	lserver, lport, _ := net.SplitHostPort(conn.LocalAddr().String())
 	ui.printMsg("[%3d] local %s port %s connected to %s port %s",
 		ec.fd, lserver, lport, rserver, rport)
-	buff := make([]byte, test.testParam.BufferSize)
+	size := test.clientParam.BufferSize
-	for i := uint32(0); i < test.testParam.BufferSize; i++ {
+	if test.clientParam.BwRate > 0 && uint64(size) > test.clientParam.BwRate {
+		size = uint32(test.clientParam.BwRate)
+	}
+	buff := make([]byte, size)
+	for i := uint32(0); i < size; i++ {
 		buff[i] = byte(i)
 	}
-	blen := len(buff)
+	start, waitTime, sendRate := beginThrottle()
 ExitForLoop:
 	for {
 		select {
@@ -243,38 +275,46 @@ ExitForLoop:
 		default:
 			n := 0
 			var err error = nil
-			if test.testParam.Reverse {
+			if test.clientParam.Reverse {
 				n, err = io.ReadFull(conn, buff)
 			} else {
 				n, err = conn.Write(buff)
 			}
-			if err != nil || n < blen {
+			if err != nil || n < int(size) {
 				ui.printDbg("Error sending/receiving data on a connection for bandwidth test: %v", err)
 				break ExitForLoop
 			}
-			atomic.AddUint64(&ec.bw, uint64(blen))
+			atomic.AddUint64(&ec.bw, uint64(size))
-			atomic.AddUint64(&test.testResult.bw, uint64(blen))
+			atomic.AddUint64(&test.testResult.bw, uint64(size))
+			sendRate += uint64(size)
+			if test.clientParam.BwRate > 0 && !test.clientParam.Reverse && sendRate >= test.clientParam.BwRate {
+				start, waitTime, sendRate = enforceThrottle(start, waitTime)
+			}
 		}
 	}
 }
 
 func runTCPLatencyTest(test *ethrTest, g time.Duration, toStop chan int) {
-	conn, err := net.Dial(tcp(ipVer), test.dialAddr)
+	ui.printMsg("Running latency test: %v, %v", test.clientParam.RttCount, test.clientParam.BufferSize)
+	conn, err := ethrDial(TCP, test.dialAddr)
 	if err != nil {
 		ui.printErr("Error dialing the latency connection: %v", err)
-		os.Exit(1)
 		return
 	}
 	defer conn.Close()
-	handshakeWithServer(test, conn)
+	err = handshakeWithServer(test, conn)
+	if err != nil {
+		ui.printErr("Failed in handshake with the server. Error: %v", err)
+		return
+	}
 	ui.emitLatencyHdr()
-	buffSize := test.testParam.BufferSize
+	buffSize := test.clientParam.BufferSize
 	buff := make([]byte, buffSize)
 	for i := uint32(0); i < buffSize; i++ {
 		buff[i] = byte(i)
 	}
 	blen := len(buff)
-	rttCount := test.testParam.RttCount
+	rttCount := test.clientParam.RttCount
 	latencyNumbers := make([]time.Duration, rttCount)
 ExitForLoop:
 	for {
@@ -342,20 +382,20 @@ func calcAndPrintLatency(test *ethrTest, rttCount uint32, latencyNumbers []time.
 	p9999 := latencyNumbers[uint64(((float64(rttCountFixed)*99.99)/100)-1)]
 	ui.emitLatencyResults(
 		test.session.remoteIP,
-		protoToString(test.testParam.TestID.Protocol),
+		protoToString(test.testID.Protocol),
 		avg, min, max, p50, p90, p95, p99, p999, p9999)
 }
 
 func tcpRunCpsTest(test *ethrTest) {
-	for th := uint32(0); th < test.testParam.NumThreads; th++ {
+	for th := uint32(0); th < test.clientParam.NumThreads; th++ {
-		go func() {
+		go func(th uint32) {
 		ExitForLoop:
 			for {
 				select {
 				case <-test.done:
 					break ExitForLoop
 				default:
-					conn, err := net.Dial(tcp(ipVer), test.dialAddr)
+					conn, err := ethrDialAll(TCP, test.dialAddr)
 					if err == nil {
 						atomic.AddUint64(&test.testResult.cps, 1)
 						tcpconn, ok := conn.(*net.TCPConn)
@@ -368,15 +408,15 @@ func tcpRunCpsTest(test *ethrTest) {
 					}
 				}
 			}
-		}()
+		}(th)
 	}
 }
 
-func clientRunPingTest(test *ethrTest, g time.Duration, warmupCount int) {
+func clientRunPingTest(test *ethrTest, g time.Duration, warmupCount uint32) {
 	// TODO: Override NumThreads for now, fix it later to support parallel
 	// threads.
-	test.testParam.NumThreads = 1
+	test.clientParam.NumThreads = 1
-	for th := uint32(0); th < test.testParam.NumThreads; th++ {
+	for th := uint32(0); th < test.clientParam.NumThreads; th++ {
 		go func() {
 			var sent, rcvd, lost uint32
 			warmupText := "[warmup] "
@@ -418,7 +458,7 @@ func clientRunPingTest(test *ethrTest, g time.Duration, warmupCount int) {
 }
 
 func clientRunPing(test *ethrTest, prefix string) (time.Duration, error) {
-	if test.testParam.TestID.Protocol == TCP {
+	if test.testID.Protocol == TCP {
 		return tcpRunPing(test, prefix)
 	} else {
 		return icmpRunPing(test, prefix)
@@ -427,8 +467,7 @@ func clientRunPing(test *ethrTest, prefix string) (time.Duration, error) {
 
 func tcpRunPing(test *ethrTest, prefix string) (timeTaken time.Duration, err error) {
 	t0 := time.Now()
-	// conn, err := net.DialTimeout(tcp(ipVer), *server, time.Second)
+	conn, err := ethrDial(TCP, test.dialAddr)
-	conn, err := net.Dial(tcp(ipVer), test.dialAddr)
 	if err != nil {
 		ui.printMsg("[tcp] %sConnection to %s: Timed out (%v)", prefix, test.dialAddr, err)
 		return
@@ -466,11 +505,6 @@ func tcpRunMyTraceRoute(test *ethrTest, gap time.Duration, toStop chan int) {
 }
 
 func tcpRunTraceRouteInternal(test *ethrTest, gap time.Duration, toStop chan int, mtrMode bool) {
-	if !IsAdmin() {
-		ui.printErr("For TCP based TraceRoute, running as administrator is required.\n")
-		toStop <- interrupt
-		return
-	}
 	gHop = make([]ethrHopData, gMaxHops)
 	err := tcpDiscoverHops(test, mtrMode)
 	if err != nil {
@@ -545,7 +579,11 @@ func tcpProbe(test *ethrTest, hop int, hopIP string, hopData *ethrHopData) (erro
 		return err, isLast
 	}
 	defer c.Close()
-	localPortNum := uint16(8888 + hop)
+	localPortNum := uint16(8888)
+	if gClientPort != 0 {
+		localPortNum = gClientPort
+	}
+	localPortNum += uint16(hop)
 	b := make([]byte, 4)
 	binary.BigEndian.PutUint16(b[0:], localPortNum)
 	remotePortNum, err := strconv.ParseUint(test.remotePort, 10, 16)
@@ -558,7 +596,12 @@ func tcpProbe(test *ethrTest, hop int, hopIP string, hopData *ethrHopData) (erro
 		peerAddrChan <- peerAddr
 	}()
 	startTime := time.Now()
-	_, err = ethrDialForTraceRoute(tcp(ipVer), test.dialAddr, localPortNum, hop)
+	conn, err := ethrDialEx(TCP, test.dialAddr, gLocalIP, localPortNum, hop, int(gTOS))
+	if err != nil {
+		ui.printDbg("Failed to Dial the connection. Error: %v", err)
+	} else {
+		conn.Close()
+	}
 	hopData.sent++
 	peerAddr := ""
 	endTime := time.Now()
@@ -741,17 +784,6 @@ ExitForLoop:
 	}
 }
 
-/*
-func icmpNewConn(localAddr string) (*icmp.PacketConn, error) {
-	c, err := icmp.ListenPacket("ip4:icmp", localAddr)
-	if err != nil {
-		ui.printErr("Failed to listen for ICMP on local address %v. Msg: %v.", localAddr, err.Error())
-		return nil, err
-	}
-	return c, nil
-}
-*/
-
 func icmpProbe(test *ethrTest, dstIPAddr net.IPAddr, icmpTimeout time.Duration, hopIP string, hopData *ethrHopData, hop, seq int) (error, bool) {
 	isLast := false
 	echoMsg := fmt.Sprintf("Hello: Ethr - %v", hop)
@@ -779,23 +811,33 @@ func icmpProbe(test *ethrTest, dstIPAddr net.IPAddr, icmpTimeout time.Duration,
 	return nil, isLast
 }
 
-const (
-	Icmpv4 = 1  // ICMP for IPv4
-	Icmpv6 = 58 // ICMP for IPv6
-)
-
 func icmpSetTTL(c net.PacketConn, ttl int) error {
 	err := os.ErrInvalid
-	if ipVer == ethrIPv4 {
+	if gIPVersion == ethrIPv4 {
 		cIPv4 := ipv4.NewPacketConn(c)
 		err = cIPv4.SetTTL(ttl)
-	} else if ipVer == ethrIPv6 {
+	} else if gIPVersion == ethrIPv6 {
 		cIPv6 := ipv6.NewPacketConn(c)
 		err = cIPv6.SetHopLimit(ttl)
 	}
 	return err
 }
 
+func icmpSetTOS(c net.PacketConn, tos int) error {
+	if tos == 0 {
+		return nil
+	}
+	err := os.ErrInvalid
+	if gIPVersion == ethrIPv4 {
+		cIPv4 := ipv4.NewPacketConn(c)
+		err = cIPv4.SetTOS(tos)
+	} else if gIPVersion == ethrIPv6 {
+		cIPv6 := ipv6.NewPacketConn(c)
+		err = cIPv6.SetTrafficClass(tos)
+	}
+	return err
+}
+
 func icmpSendMsg(c net.PacketConn, dstIPAddr net.IPAddr, hop, seq int, body string, timeout time.Duration) (time.Time, []byte, error) {
 	start := time.Now()
 	err := icmpSetTTL(c, hop+1)
@@ -803,6 +845,7 @@ func icmpSendMsg(c net.PacketConn, dstIPAddr net.IPAddr, hop, seq int, body stri
 		ui.printErr("Failed to set TTL. Error: %v", err)
 		return start, nil, err
 	}
+	icmpSetTOS(c, int(gTOS))
 
 	err = c.SetDeadline(time.Now().Add(timeout))
 	if err != nil {
@@ -819,7 +862,7 @@ func icmpSendMsg(c net.PacketConn, dstIPAddr net.IPAddr, hop, seq int, body stri
 			Data: []byte(body),
 		},
 	}
-	if ipVer == ethrIPv6 {
+	if gIPVersion == ethrIPv6 {
 		wm.Type = ipv6.ICMPTypeEchoRequest
 	}
 	wb, err := wm.Marshal(nil)
@@ -865,7 +908,7 @@ func icmpRecvMsg(c net.PacketConn, proto EthrProtocol, timeout time.Duration, ne
 			ui.printDbg("Matching peer is not found.")
 			continue
 		}
-		icmpMsg, err := icmp.ParseMessage(IcmpProto(ipVer), b[:n])
+		icmpMsg, err := icmp.ParseMessage(IcmpProto(), b[:n])
 		if err != nil {
 			ui.printDbg("Failed to parse ICMP message: %v", err)
 			continue
@@ -875,13 +918,14 @@ func icmpRecvMsg(c net.PacketConn, proto EthrProtocol, timeout time.Duration, ne
 			index := bytes.Index(body, neededSig[:4])
 			if index > 0 {
 				if proto == TCP {
+					ui.printDbg("Found correct ICMP error message. PeerAddr: %v", peerAddr)
 					return peerAddr, isLast, nil
 				} else if proto == ICMP {
 					if index < 4 {
 						ui.printDbg("Incorrect length of ICMP message.")
 						continue
 					}
-					innerIcmpMsg, _ := icmp.ParseMessage(IcmpProto(ipVer), body[index-4:])
+					innerIcmpMsg, _ := icmp.ParseMessage(IcmpProto(), body[index-4:])
 					switch innerIcmpMsg.Body.(type) {
 					case *icmp.Echo:
 						seq := innerIcmpMsg.Body.(*icmp.Echo).Seq
@@ -912,10 +956,14 @@ func icmpRecvMsg(c net.PacketConn, proto EthrProtocol, timeout time.Duration, ne
 }
 
 func runUDPBandwidthAndPpsTest(test *ethrTest) {
-	for th := uint32(0); th < test.testParam.NumThreads; th++ {
+	for th := uint32(0); th < test.clientParam.NumThreads; th++ {
-		go func() {
+		go func(th uint32) {
-			buff := make([]byte, test.testParam.BufferSize)
+			size := test.clientParam.BufferSize
-			conn, err := net.Dial(udp(ipVer), test.dialAddr)
+			if test.clientParam.BwRate > 0 && uint64(size) > test.clientParam.BwRate {
+				size = uint32(test.clientParam.BwRate)
+			}
+			buff := make([]byte, size)
+			conn, err := ethrDialInc(UDP, test.dialAddr, uint16(th))
 			if err != nil {
 				ui.printDbg("Unable to dial UDP, error: %v", err)
 				return
@@ -927,6 +975,7 @@ func runUDPBandwidthAndPpsTest(test *ethrTest) {
 			ui.printMsg("[%3d] local %s port %s connected to %s port %s",
 				ec.fd, lserver, lport, rserver, rport)
 			blen := len(buff)
+			start, waitTime, sendRate := beginThrottle()
 		ExitForLoop:
 			for {
 				select {
@@ -946,84 +995,12 @@ func runUDPBandwidthAndPpsTest(test *ethrTest) {
 					atomic.AddUint64(&ec.pps, 1)
 					atomic.AddUint64(&test.testResult.bw, uint64(n))
 					atomic.AddUint64(&test.testResult.pps, 1)
+					sendRate += uint64(size)
+					if test.clientParam.BwRate > 0 && !test.clientParam.Reverse && sendRate >= test.clientParam.BwRate {
+						start, waitTime, sendRate = enforceThrottle(start, waitTime)
+					}
 				}
 			}
-		}()
+		}(th)
 	}
 }
-
-/*
-func runHTTPBandwidthTest(test *ethrTest) {
-	uri := test.session.remoteIP
-	ui.printMsg("uri=%s", uri)
-	uri = "http://" + uri + ":" + httpBandwidthPort
-	for th := uint32(0); th < test.testParam.NumThreads; th++ {
-		buff := make([]byte, test.testParam.BufferSize)
-		for i := uint32(0); i < test.testParam.BufferSize; i++ {
-			buff[i] = 'x'
-		}
-		tr := &http.Transport{DisableCompression: true}
-		client := &http.Client{Transport: tr}
-		go runHTTPandHTTPSBandwidthTest(test, client, uri, buff)
-	}
-}
-
-func runHTTPSBandwidthTest(test *ethrTest) {
-	uri := test.session.remoteIP
-	uri = "https://" + uri + ":" + httpsBandwidthPort
-	for th := uint32(0); th < test.testParam.NumThreads; th++ {
-		buff := make([]byte, test.testParam.BufferSize)
-		for i := uint32(0); i < test.testParam.BufferSize; i++ {
-			buff[i] = 'x'
-		}
-		c, err := x509.ParseCertificate(gCert)
-		if err != nil {
-			ui.printErr("runHTTPSBandwidthTest: failed to parse certificate: %v", err)
-		}
-		clientCertPool := x509.NewCertPool()
-		clientCertPool.AddCert(c)
-
-		tlsConfig := &tls.Config{
-			InsecureSkipVerify: gIgnoreCert,
-			// Certificates: []tls.Certificate{cert},
-			RootCAs: clientCertPool,
-		}
-		//tlsConfig.BuildNameToCertificate()
-		tr := &http.Transport{DisableCompression: true, TLSClientConfig: tlsConfig}
-		client := &http.Client{Transport: tr}
-		go runHTTPandHTTPSBandwidthTest(test, client, uri, buff)
-	}
-}
-
-func runHTTPandHTTPSBandwidthTest(test *ethrTest, client *http.Client, uri string, buff []byte) {
-ExitForLoop:
-	for {
-		select {
-		case <-test.done:
-			break ExitForLoop
-		default:
-			// response, err := http.Get(uri)
-			response, err := client.Post(uri, "text/plain", bytes.NewBuffer(buff))
-			if err != nil {
-				ui.printDbg("Error in HTTP request: %v", err)
-				continue
-			} else {
-				ui.printDbg("Status received: %v", response.StatusCode)
-				if response.StatusCode != http.StatusOK {
-					ui.printDbg("Error in HTTP request, received status: %v", response.StatusCode)
-					continue
-				}
-				contents, err := ioutil.ReadAll(response.Body)
-				response.Body.Close()
-				if err != nil {
-					ui.printDbg("Error in receving HTTP response: %v", err)
-					continue
-				}
-				ethrUnused(contents)
-				// ui.printDbg("%s", string(contents))
-			}
-			atomic.AddUint64(&test.testResult.data, uint64(test.testParam.BufferSize))
-		}
-	}
-}
-*/

clientui.go

@@ -111,11 +111,11 @@ func printBwTestResult(p EthrProtocol, fd string, t0, t1, bw, pps uint64) {
 }
 
 func printTestResult(test *ethrTest, seconds uint64) {
-	if test.testParam.TestID.Type == Bandwidth && (test.testParam.TestID.Protocol == TCP ||
+	if test.testID.Type == Bandwidth &&
-		test.testParam.TestID.Protocol == UDP) {
+		(test.testID.Protocol == TCP || test.testID.Protocol == UDP) {
 		if gInterval == 0 {
-			printBwTestDivider(test.testParam.TestID.Protocol)
+			printBwTestDivider(test.testID.Protocol)
-			printBwTestHeader(test.testParam.TestID.Protocol)
+			printBwTestHeader(test.testID.Protocol)
 		}
 		cbw := uint64(0)
 		cpps := uint64(0)
@@ -126,32 +126,32 @@ func printTestResult(test *ethrTest, seconds uint64) {
 			bw /= seconds
 			if !gNoConnectionStats {
 				fd := fmt.Sprintf("%5d", ec.fd)
-				printBwTestResult(test.testParam.TestID.Protocol, fd, gInterval, gInterval+1, bw, pps)
+				printBwTestResult(test.testID.Protocol, fd, gInterval, gInterval+1, bw, pps)
 			}
 			cbw += bw
 			cpps += pps
 			ccount++
 		})
 		if ccount > 1 || gNoConnectionStats {
-			printBwTestResult(test.testParam.TestID.Protocol, "SUM", gInterval, gInterval+1, cbw, cpps)
+			printBwTestResult(test.testID.Protocol, "SUM", gInterval, gInterval+1, cbw, cpps)
 			if !gNoConnectionStats {
-				printBwTestDivider(test.testParam.TestID.Protocol)
+				printBwTestDivider(test.testID.Protocol)
 			}
 		}
-		logResults([]string{test.session.remoteIP, protoToString(test.testParam.TestID.Protocol),
+		logResults([]string{test.session.remoteIP, protoToString(test.testID.Protocol),
 			bytesToRate(cbw), "", ppsToString(cpps), ""})
-	} else if test.testParam.TestID.Type == Cps {
+	} else if test.testID.Type == Cps {
 		if gInterval == 0 {
 			ui.printMsg("- - - - - - - - - - - - - - - - - - ")
 			ui.printMsg("Protocol    Interval      Conn/s")
 		}
 		cps := atomic.SwapUint64(&test.testResult.cps, 0)
 		ui.printMsg("  %-5s    %03d-%03d sec   %7s",
-			protoToString(test.testParam.TestID.Protocol),
+			protoToString(test.testID.Protocol),
 			gInterval, gInterval+1, cpsToString(cps))
-		logResults([]string{test.session.remoteIP, protoToString(test.testParam.TestID.Protocol),
+		logResults([]string{test.session.remoteIP, protoToString(test.testID.Protocol),
 			"", cpsToString(cps), "", ""})
-	} else if test.testParam.TestID.Type == Pps {
+	} else if test.testID.Type == Pps {
 		if gInterval == 0 {
 			ui.printMsg("- - - - - - - - - - - - - - - - - - - - - - -")
 			ui.printMsg("Protocol    Interval      Bits/s    Pkts/s")
@@ -159,11 +159,11 @@ func printTestResult(test *ethrTest, seconds uint64) {
 		bw := atomic.SwapUint64(&test.testResult.bw, 0)
 		pps := atomic.SwapUint64(&test.testResult.pps, 0)
 		ui.printMsg("  %-5s    %03d-%03d sec   %7s   %7s",
-			protoToString(test.testParam.TestID.Protocol),
+			protoToString(test.testID.Protocol),
 			gInterval, gInterval+1, bytesToRate(bw), ppsToString(pps))
-		logResults([]string{test.session.remoteIP, protoToString(test.testParam.TestID.Protocol),
+		logResults([]string{test.session.remoteIP, protoToString(test.testID.Protocol),
 			bytesToRate(bw), "", ppsToString(pps), ""})
-	} else if test.testParam.TestID.Type == MyTraceRoute {
+	} else if test.testID.Type == MyTraceRoute {
 		if gCurHops > 0 {
 			ui.printMsg("- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ")
 			ui.printMsg("Host: %-40s    Sent    Recv        Last         Avg        Best        Wrst", test.session.remoteIP)
@@ -188,14 +188,11 @@ func printTestResult(test *ethrTest, seconds uint64) {
 }
 
 func (u *clientUI) emitTestResult(s *ethrSession, proto EthrProtocol, seconds uint64) {
-	//var data uint64
 	var testList = []EthrTestType{Bandwidth, Cps, Pps, TraceRoute, MyTraceRoute}
 
 	for _, testType := range testList {
 		test, found := s.tests[EthrTestID{proto, testType}]
 		if found && test.isActive {
-			//data = atomic.SwapUint64(&test.testResult.data, 0)
-			//data /= seconds
 			printTestResult(test, seconds)
 		}
 	}

ethr.go

@@ -8,6 +8,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"net"
 	"os"
 	"runtime"
 	"strings"
@@ -43,91 +44,226 @@ func main() {
 	//
 	runtime.GOMAXPROCS(1024)
 
+	// Common
 	flag.Usage = func() { ethrUsage() }
-	isServer := flag.Bool("s", false, "")
+	noOutput := flag.Bool("no", false, "")
-	clientDest := flag.String("c", "", "")
-	testTypePtr := flag.String("t", "", "")
-	thCount := flag.Int("n", 1, "")
-	bufLenStr := flag.String("l", "", "")
-	protocol := flag.String("p", "tcp", "")
 	outputFile := flag.String("o", defaultLogFileName, "")
 	debug := flag.Bool("debug", false, "")
-	noOutput := flag.Bool("no", false, "")
-	duration := flag.Duration("d", 10*time.Second, "")
-	showUI := flag.Bool("ui", false, "")
-	rttCount := flag.Int("i", 1000, "")
-	port := flag.Int("port", 8888, "")
-	modeStr := flag.String("m", "", "")
 	use4 := flag.Bool("4", false, "")
 	use6 := flag.Bool("6", false, "")
+	port := flag.Int("port", 8888, "")
+	ip := flag.String("ip", "", "")
+	// Server
+	isServer := flag.Bool("s", false, "")
+	showUI := flag.Bool("ui", false, "")
+	// Client & External Client
+	clientDest := flag.String("c", "", "")
+	bufLenStr := flag.String("l", "", "")
+	bwRateStr := flag.String("b", "", "")
+	cport := flag.Int("cport", 0, "")
+	duration := flag.Duration("d", 10*time.Second, "")
 	gap := flag.Duration("g", time.Second, "")
-	reverse := flag.Bool("r", false, "")
+	iterCount := flag.Int("i", 1000, "")
 	ncs := flag.Bool("ncs", false, "")
-	ic := flag.Bool("ic", false, "")
+	protocol := flag.String("p", "tcp", "")
+	reverse := flag.Bool("r", false, "")
+	testTypePtr := flag.String("t", "", "")
+	tos := flag.Int("tos", 0, "")
+	thCount := flag.Int("n", 1, "")
 	wc := flag.Int("w", 1, "")
+	xClientDest := flag.String("x", "", "")
 
 	flag.Parse()
 
-	//
-	// TODO: Handle the case if there are incorrect arguments
-	// fmt.Println("Number of incorrect arguments: " + strconv.Itoa(flag.NArg()))
-	//
-
-	//
-	// Only used in client mode, to control whether to display per connection
-	// statistics or not.
-	//
-	gNoConnectionStats = *ncs
-
-	//
-	// Only used in client mode to ignore HTTPS cert errors.
-	//
-	gIgnoreCert = *ic
-
-	if *debug {
-		loggingLevel = LogLevelDebug
-	}
-
-	xMode = false
-	switch *modeStr {
-	case "":
-	case "x":
-		xMode = true
-	default:
-		printUsageError("Invalid value for client mode (-m).")
-	}
-
 	if *isServer {
 		if *clientDest != "" {
 			printUsageError("Invalid arguments, \"-c\" cannot be used with \"-s\".")
 		}
+		if *xClientDest != "" {
+			printUsageError("Invalid arguments, \"-x\" cannot be used with \"-s\".")
+		}
+		if *bufLenStr != "" {
+			printServerModeArgError("l")
+		}
+		if *bwRateStr != "" {
+			printServerModeArgError("b")
+		}
+		if *cport != 0 {
+			printServerModeArgError("cport")
+		}
+		if *duration != 10*time.Second {
+			printServerModeArgError("d")
+		}
+		if *gap != time.Second {
+			printServerModeArgError("g")
+		}
+		if *iterCount != 1000 {
+			printServerModeArgError("i")
+		}
+		if *ncs {
+			printServerModeArgError("ncs")
+		}
+		if *protocol != "tcp" {
+			printServerModeArgError("p")
+		}
 		if *reverse {
-			printUsageError("Invalid arguments, \"-r\" can only be used in client (\"-c\") mode.")
+			printServerModeArgError("r")
 		}
-		if xMode {
+		if *testTypePtr != "" {
-			printUsageError("Invalid argument, \"-m\" can only be used in client (\"-c\") mode.")
+			printServerModeArgError("t")
 		}
-	} else if *clientDest == "" {
+		if *tos != 0 {
+			printServerModeArgError("tos")
+		}
+		if *thCount != 1 {
+			printServerModeArgError("n")
+		}
+		if *wc != 1 {
+			printServerModeArgError("wc")
+		}
+	} else if *clientDest != "" || *xClientDest != "" {
+		if *clientDest != "" && *xClientDest != "" {
+			printUsageError("Invalid argument, both \"-c\" and \"-x\" cannot be specified at the same time.")
+		}
+		if *showUI {
+			printUsageError(fmt.Sprintf("Invalid argument, \"-%s\" can only be used in server (\"-s\") mode.", "ui"))
+		}
+	} else {
 		printUsageError("Invalid arguments, use either \"-s\" or \"-c\".")
 	}
 
+	// Process common parameters.
+
+	if *debug {
+		loggingLevel = LogLevelDebug
+	}
+
 	if *use4 && !*use6 {
-		ipVer = ethrIPv4
+		gIPVersion = ethrIPv4
 	} else if *use6 && !*use4 {
-		ipVer = ethrIPv6
+		gIPVersion = ethrIPv6
+	}
+
+	if *ip != "" {
+		gLocalIP = *ip
+		ipAddr := net.ParseIP(gLocalIP)
+		if ipAddr == nil {
+			printUsageError(fmt.Sprintf("Invalid IP address: <%s> specified.", *ip))
+		}
+		if (gIPVersion == ethrIPv4 && ipAddr.To4() == nil) || (gIPVersion == ethrIPv6 && ipAddr.To16() == nil) {
+			printUsageError(fmt.Sprintf("Invalid IP address version: <%s> specified.", *ip))
+		}
+	}
+	gEthrPort = uint16(*port)
+	gEthrPortStr = fmt.Sprintf("%d", gEthrPort)
+
+	logFileName := *outputFile
+	if !*noOutput {
+		if logFileName == defaultLogFileName {
+			if *isServer {
+				logFileName = "ethrs.log"
+			} else {
+				logFileName = "ethrc.log"
+			}
+		}
+		logInit(logFileName)
 	}
 
 	var testType EthrTestType
-	switch *testTypePtr {
+	var destination string
+	if *isServer {
+		// Server side parameter processing.
+		testType = All
+		serverParam := ethrServerParam{*showUI}
+		runServer(serverParam)
+	} else {
+		gIsExternalClient = false
+		destination = *clientDest
+		if *xClientDest != "" {
+			gIsExternalClient = true
+			destination = *xClientDest
+		}
+		gNoConnectionStats = *ncs
+		testType = getTestType(*testTypePtr)
+		proto := getProtocol(*protocol)
+
+		// Default latency test to 1B if length is not specified
+		switch *bufLenStr {
+		case "":
+			*bufLenStr = getDefaultBufferLenStr(*testTypePtr)
+		}
+		bufLen := unitToNumber(*bufLenStr)
+		if bufLen == 0 {
+			printUsageError(fmt.Sprintf("Invalid length specified: %s" + *bufLenStr))
+		}
+
+		// Check specific bwRate if any.
+		bwRate := uint64(0)
+		if *bwRateStr != "" {
+			bwRate = unitToNumber(*bwRateStr)
+		}
+
+		//
+		// For Pkt/s, we always override the buffer size to be just 1 byte.
+		// TODO: Evaluate in future, if we need to support > 1 byte packets for
+		//       Pkt/s testing.
+		//
+		if testType == Pps {
+			bufLen = 1
+		}
+
+		if *iterCount <= 0 {
+			printUsageError(fmt.Sprintf("Invalid iteration count for latency test: %d", *iterCount))
+		}
+
+		if *thCount <= 0 {
+			*thCount = runtime.NumCPU()
+		}
+
+		gClientPort = uint16(*cport)
+
+		testId := EthrTestID{EthrProtocol(proto), testType}
+		clientParam := EthrClientParam{
+			uint32(*thCount),
+			uint32(bufLen),
+			uint32(*iterCount),
+			*reverse,
+			*duration,
+			*gap,
+			uint32(*wc),
+			uint64(bwRate),
+			uint8(*tos)}
+		validateClientParams(testId, clientParam)
+
+		rServer := destination
+		runClient(testId, clientParam, rServer)
+	}
+}
+
+func getProtocol(protoStr string) (proto EthrProtocol) {
+	p := strings.ToUpper(protoStr)
+	proto = TCP
+	switch p {
+	case "TCP":
+		proto = TCP
+	case "UDP":
+		proto = UDP
+	case "ICMP":
+		proto = ICMP
+	default:
+		printUsageError(fmt.Sprintf("Invalid value \"%s\" specified for parameter \"-p\".\n"+
+			"Valid parameters and values are:\n", protoStr))
+	}
+	return
+}
+
+func getTestType(testTypeStr string) (testType EthrTestType) {
+	switch testTypeStr {
 	case "":
-		if *isServer {
+		if gIsExternalClient {
-			testType = All
+			testType = Ping
 		} else {
-			if xMode {
+			testType = Bandwidth
-				testType = Ping
-			} else {
-				testType = Bandwidth
-			}
 		}
 	case "b":
 		testType = Bandwidth
@@ -145,93 +281,9 @@ func main() {
 		testType = MyTraceRoute
 	default:
 		printUsageError(fmt.Sprintf("Invalid value \"%s\" specified for parameter \"-t\".\n"+
-			"Valid parameters and values are:\n", *testTypePtr))
+			"Valid parameters and values are:\n", testTypeStr))
-	}
-
-	// Default latency test to 1B if length is not specified
-	switch *bufLenStr {
-	case "":
-		*bufLenStr = getDefaultBufferLenStr(*testTypePtr)
-	}
-
-	bufLen := unitToNumber(*bufLenStr)
-	if bufLen == 0 {
-		printUsageError(fmt.Sprintf("Invalid length specified: %s" + *bufLenStr))
-	}
-
-	//
-	// For Pkt/s, we always override the buffer size to be just 1 byte.
-	// TODO: Evaluate in future, if we need to support > 1 byte packets for
-	//       Pkt/s testing.
-	//
-	if testType == Pps {
-		bufLen = 1
-	}
-
-	if *rttCount <= 0 {
-		printUsageError(fmt.Sprintf("Invalid RTT count for latency test: %d", *rttCount))
-	}
-
-	p := strings.ToUpper(*protocol)
-	proto := TCP
-	switch p {
-	case "TCP":
-		proto = TCP
-	case "UDP":
-		proto = UDP
-	case "HTTP":
-		proto = HTTP
-	case "HTTPS":
-		proto = HTTPS
-	case "ICMP":
-		proto = ICMP
-	default:
-		printUsageError(fmt.Sprintf("Invalid value \"%s\" specified for parameter \"-p\".\n"+
-			"Valid parameters and values are:\n", *protocol))
-	}
-
-	// Override ipVer because for ICMP, specific version is required.
-	if proto == ICMP || (testType == TraceRoute || testType == MyTraceRoute) {
-		if ipVer == ethrIPAny {
-			ipVer = ethrIPv4
-		}
-	}
-
-	if *thCount <= 0 {
-		*thCount = runtime.NumCPU()
-	}
-
-	testParam := EthrTestParam{EthrTestID{EthrProtocol(proto), testType},
-		uint32(*thCount),
-		uint32(bufLen),
-		uint32(*rttCount),
-		*reverse}
-	validateTestParam(*isServer, testParam)
-
-	gEthrPort = *port
-	gEthrPortStr = fmt.Sprintf("%d", gEthrPort)
-
-	logFileName := *outputFile
-	if !*noOutput {
-		if logFileName == defaultLogFileName {
-			if *isServer {
-				logFileName = "ethrs.log"
-			} else {
-				logFileName = "ethrc.log"
-			}
-		}
-		logInit(logFileName)
-	}
-
-	clientParam := ethrClientParam{*duration, *gap, *wc}
-	serverParam := ethrServerParam{*showUI}
-
-	if *isServer {
-		runServer(testParam, serverParam)
-	} else {
-		rServer := *clientDest
-		runClient(testParam, clientParam, rServer)
 	}
+	return
 }
 
 func getDefaultBufferLenStr(testTypePtr string) string {
@@ -241,80 +293,82 @@ func getDefaultBufferLenStr(testTypePtr string) string {
 	return defaultBufferLenStr
 }
 
-func emitUnsupportedTest(testParam EthrTestParam) {
+func validateClientParams(testID EthrTestID, clientParam EthrClientParam) {
+	if !gIsExternalClient {
+		validateClientTest(testID, clientParam)
+	} else {
+		validateExtModeClientTest(testID)
+	}
+}
+
+func validateClientTest(testID EthrTestID, clientParam EthrClientParam) {
+	testType := testID.Type
+	protocol := testID.Protocol
+	switch protocol {
+	case TCP:
+		if testType != Bandwidth && testType != Cps && testType != Latency && testType != Ping && testType != TraceRoute && testType != MyTraceRoute {
+			emitUnsupportedTest(testID)
+		}
+		if clientParam.Reverse && testType != Bandwidth {
+			printReverseModeError()
+		}
+		if clientParam.BufferSize > 2*GIGA {
+			printUsageError("Maximum allowed value for \"-l\" for TCP is 2GB.")
+		}
+	case UDP:
+		if testType != Bandwidth && testType != Pps {
+			emitUnsupportedTest(testID)
+		}
+		if testType == Bandwidth {
+			if clientParam.BufferSize > (64 * 1024) {
+				printUsageError("Maximum supported buffer size for UDP is 64K\n")
+			}
+		}
+		if clientParam.Reverse {
+			printReverseModeError()
+		}
+		if clientParam.BufferSize > 64*KILO {
+			printUsageError("Maximum allowed value for \"-l\" for TCP is 64KB.")
+		}
+	default:
+		emitUnsupportedTest(testID)
+	}
+}
+
+func validateExtModeClientTest(testID EthrTestID) {
+	testType := testID.Type
+	protocol := testID.Protocol
+	switch protocol {
+	case TCP:
+		if testType != Ping && testType != Cps && testType != TraceRoute && testType != MyTraceRoute {
+			emitUnsupportedTest(testID)
+		}
+	case ICMP:
+		if testType != Ping && testType != TraceRoute && testType != MyTraceRoute {
+			emitUnsupportedTest(testID)
+		}
+	default:
+		emitUnsupportedTest(testID)
+	}
+}
+
+func printServerModeArgError(arg string) {
+	printUsageError(fmt.Sprintf("Invalid argument, \"-%s\" can only be used in client (\"-c\") mode.", arg))
+}
+
+func emitUnsupportedTest(testID EthrTestID) {
 	printUsageError(fmt.Sprintf("Test: \"%s\" for Protocol: \"%s\" is not supported.\n",
-		testToString(testParam.TestID.Type), protoToString(testParam.TestID.Protocol)))
+		testToString(testID.Type), protoToString(testID.Protocol)))
 }
 
 func printReverseModeError() {
 	printUsageError("Reverse mode (-r) is only supported for TCP Bandwidth tests.")
 }
 
-func validateTestParam(isServer bool, testParam EthrTestParam) {
+func printUsageError(s string) {
-	testType := testParam.TestID.Type
+	fmt.Printf("Error: %s\n", s)
-	protocol := testParam.TestID.Protocol
+	fmt.Printf("Please use \"ethr -h\" for complete list of command line arguments.\n")
-	if isServer {
+	os.Exit(1)
-		if testType != All || protocol != TCP {
-			emitUnsupportedTest(testParam)
-		}
-	} else {
-		if !xMode {
-			switch protocol {
-			case TCP:
-				if testType != Bandwidth && testType != Cps && testType != Latency && testType != Ping && testType != TraceRoute && testType != MyTraceRoute {
-					emitUnsupportedTest(testParam)
-				}
-				if testParam.Reverse && testType != Bandwidth {
-					printReverseModeError()
-				}
-			case UDP:
-				if testType != Bandwidth && testType != Pps {
-					emitUnsupportedTest(testParam)
-				}
-				if testType == Bandwidth {
-					if testParam.BufferSize > (64 * 1024) {
-						printUsageError("Maximum supported buffer size for UDP is 64K\n")
-					}
-				}
-				if testParam.Reverse {
-					printReverseModeError()
-				}
-			case ICMP:
-				if testType != TraceRoute && testType != MyTraceRoute {
-					emitUnsupportedTest(testParam)
-				}
-			case HTTP:
-				if testType != Bandwidth && testType != Latency {
-					emitUnsupportedTest(testParam)
-				}
-				if testParam.Reverse {
-					printReverseModeError()
-				}
-			case HTTPS:
-				if testType != Bandwidth {
-					emitUnsupportedTest(testParam)
-				}
-				if testParam.Reverse {
-					printReverseModeError()
-				}
-			default:
-				emitUnsupportedTest(testParam)
-			}
-		} else {
-			switch protocol {
-			case TCP:
-				if testType != Ping && testType != Cps && testType != TraceRoute && testType != MyTraceRoute {
-					emitUnsupportedTest(testParam)
-				}
-			case ICMP:
-				if testType != Ping && testType != TraceRoute && testType != MyTraceRoute {
-					emitUnsupportedTest(testParam)
-				}
-			default:
-				emitUnsupportedTest(testParam)
-			}
-		}
-	}
 }
 
 // ethrUsage prints the command-line usage text
@@ -337,35 +391,42 @@ func ethrUsage() {
 	fmt.Println("In this mode, Ethr runs as a server, allowing multiple clients to run")
 	fmt.Println("performance tests against it.")
 	printServerUsage()
-	printFlagUsage("ui", "", "Show output in text UI.")
+	printIPUsage()
 	printPortUsage()
+	printFlagUsage("ui", "", "Show output in text UI.")
 
 	fmt.Println("\nMode: Client")
 	fmt.Println("================================================================================")
 	fmt.Println("In this mode, Ethr client can only talk to an Ethr server.")
 	printClientUsage()
+	printBwRateUsage()
+	printCPortUsage()
 	printDurationUsage()
 	printGapUsage()
 	printIterationUsage()
+	printIPUsage()
 	printBufLenUsage()
 	printThreadUsage()
 	printProtocolUsage()
 	printPortUsage()
 	printFlagUsage("r", "", "For Bandwidth tests, send data from server to client.")
 	printTestType()
+	printToSUsage()
 	printWarmupUsage()
 
-	fmt.Println("\nMode: External Client")
+	fmt.Println("\nMode: External")
 	fmt.Println("================================================================================")
-	fmt.Println("In this mode, Ethr client can talk to a non-Ethr server. This mode only supports")
+	fmt.Println("In this mode, Ethr talks to a non-Ethr server. This mode supports only a")
 	fmt.Println("few types of measurements, such as Ping, Connections/s and TraceRoute.")
 	printExtClientUsage()
-	printModeUsage()
+	printCPortUsage()
 	printDurationUsage()
 	printGapUsage()
+	printIPUsage()
 	printThreadUsage()
 	printExtProtocolUsage()
 	printExtTestType()
+	printToSUsage()
 	printWarmupUsage()
 }
 
@@ -386,9 +447,10 @@ func printClientUsage() {
 }
 
 func printExtClientUsage() {
-	printFlagUsage("c", "<destination>", "Run in external client mode and connect to <destination>.",
+	printFlagUsage("x", "<destination>", "Run in external client mode and connect to <destination>.",
-		"<destination> is specified in <host:port> format for TCP and <host> format for ICMP.",
+		"<destination> is specified in URL or Host:Port format.",
-		"Example: For TCP - www.microsoft.com:443 or 10.1.0.4:22",
+		"For URL, if port is not specified, it is assumed to be 80 for http and 443 for https.",
+		"Example: For TCP - www.microsoft.com:443 or 10.1.0.4:22 or https://www.github.com",
 		"         For ICMP - www.microsoft.com or 10.1.0.4")
 }
 
@@ -465,11 +527,6 @@ func printIterationUsage() {
 		"Default: 1000")
 }
 
-func printModeUsage() {
-	printFlagUsage("m", "<mode>",
-		"'-m x' MUST be specified for external mode.")
-}
-
 func printNoConnStatUsage() {
 	printFlagUsage("ncs", "",
 		"No per Connection Stats would be printed if this flag is specified.",
@@ -488,8 +545,25 @@ func printWarmupUsage() {
 		"Default: 1")
 }
 
-func printUsageError(s string) {
+func printToSUsage() {
-	fmt.Printf("Error: %s\n", s)
+	printFlagUsage("tos", "",
-	fmt.Printf("Please use \"ethr -h\" for complete list of command line arguments.\n")
+		"Specifies 8-bit value to use in IPv4 TOS field or IPv6 Traffic Class field.")
-	os.Exit(1)
+}
+
+func printBwRateUsage() {
+	printFlagUsage("b", "<rate>",
+		"Bytes to send per second (format: <num>[KB | MB | GB])",
+		"Only valid for Bandwidth tests. Default: 0 - Unlimited",
+		"Examples: 100 (100B/s or 800bits/s), 1MB (1MB/s or 8Mbits/s).")
+}
+
+func printCPortUsage() {
+	printFlagUsage("cport", "<number>", "Use specified local port number in client for TCP & UDP tests.",
+		"Default: 0 - Ephemeral Port")
+}
+
+func printIPUsage() {
+	printFlagUsage("ip", "<string>", "Bind to specified local IP address for TCP & UDP tests.",
+		"This must be a valid IPv4 or IPv6 address.",
+		"Default: <empty> - Any IP")
 }

log.go

@@ -64,7 +64,7 @@ func logInit(fileName string) {
 	}
 	logFile, err := os.OpenFile(fileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0666)
 	if err != nil {
-		fmt.Printf("Unable to open the log file %s, Error: %v", fileName, err)
+		fmt.Printf("Unable to open the log file %s, Error: %v\n", fileName, err)
 		return
 	}
 	log.SetFlags(0)

plt_darwin.go

@@ -369,19 +369,34 @@ func setSockOptInt(fd uintptr, level, opt, val int) (err error) {
 }
 
 func IcmpNewConn(address string) (net.PacketConn, error) {
-	dialedConn, err := net.Dial(Icmp(ipVer), address)
+	dialedConn, err := net.Dial(Icmp(), address)
 	if err != nil {
 		return nil, err
 	}
 	localAddr := dialedConn.LocalAddr()
 	dialedConn.Close()
-	conn, err := net.ListenPacket(Icmp(ipVer), localAddr.String())
+	conn, err := net.ListenPacket(Icmp(), localAddr.String())
 	if err != nil {
 		return nil, err
 	}
 	return conn, nil
 }
 
+func VerifyPermissionForTest(testID EthrTestID) {
+	if testID.Protocol == ICMP || (testID.Protocol == TCP &&
+		(testID.Type == TraceRoute || testID.Type == MyTraceRoute)) {
+		if !IsAdmin() {
+			ui.printMsg("Warning: You are not running as administrator. For %s based %s",
+				protoToString(testID.Protocol), testToString(testID.Type))
+			ui.printMsg("test, running as administrator is required.\n")
+		}
+	}
+}
+
 func IsAdmin() bool {
 	return true
 }
+
+func SetTClass(fd uintptr, tos int) {
+	setSockOptInt(fd, syscall.IPPROTO_IPV6, syscall.IPV6_TCLASS, tos)
+}

plt_linux.go

@@ -162,19 +162,34 @@ func setSockOptInt(fd uintptr, level, opt, val int) (err error) {
 }
 
 func IcmpNewConn(address string) (net.PacketConn, error) {
-	dialedConn, err := net.Dial(Icmp(ipVer), address)
+	dialedConn, err := net.Dial(Icmp(), address)
 	if err != nil {
 		return nil, err
 	}
 	localAddr := dialedConn.LocalAddr()
 	dialedConn.Close()
-	conn, err := net.ListenPacket(Icmp(ipVer), localAddr.String())
+	conn, err := net.ListenPacket(Icmp(), localAddr.String())
 	if err != nil {
 		return nil, err
 	}
 	return conn, nil
 }
 
-func IsAdmin() bool {
+func VerifyPermissionForTest(testID EthrTestID) {
-	return true
+	if testID.Protocol == ICMP || (testID.Protocol == TCP &&
+		(testID.Type == TraceRoute || testID.Type == MyTraceRoute)) {
+		if !IsAdmin() {
+			ui.printMsg("Warning: You are not running as administrator. For %s based %s",
+				protoToString(testID.Protocol), testToString(testID.Type))
+			ui.printMsg("test, running as administrator is required.\n")
+		}
+	}
+}
+
+func IsAdmin() bool {
+	return os.Geteuid() == 0
+}
+
+func SetTClass(fd uintptr, tos int) {
+	setSockOptInt(fd, syscall.IPPROTO_IPV6, syscall.IPV6_TCLASS, tos)
 }

plt_windows.go

@@ -230,7 +230,7 @@ func IcmpNewConn(address string) (net.PacketConn, error) {
 	// https://github.com/golang/go/issues/38427
 
 	// First, get the correct local interface address, as SIO_RCVALL can't be set on a 0.0.0.0 listeners.
-	dialedConn, err := net.Dial(Icmp(ipVer), address)
+	dialedConn, err := net.Dial(Icmp(), address)
 	if err != nil {
 		return nil, err
 	}
@@ -248,7 +248,7 @@ func IcmpNewConn(address string) (net.PacketConn, error) {
 	}
 
 	// Bind to interface.
-	conn, err := cfg.ListenPacket(context.Background(), Icmp(ipVer), localAddr.String())
+	conn, err := cfg.ListenPacket(context.Background(), Icmp(), localAddr.String())
 	if err != nil {
 		return nil, err
 	}
@@ -261,17 +261,33 @@ func IcmpNewConn(address string) (net.PacketConn, error) {
 	size := uint32(unsafe.Sizeof(flag))
 	err = syscall.WSAIoctl(socketHandle, SIO_RCVALL, (*byte)(unsafe.Pointer(&flag)), size, nil, 0, &unused, nil, 0)
 	if err != nil {
-		return nil, err
+		// Ignore the error as for ICMP related TraceRoute, this is not required.
 	}
 
 	return conn, nil
 }
 
+func VerifyPermissionForTest(testID EthrTestID) {
+	if (testID.Type == TraceRoute || testID.Type == MyTraceRoute) &&
+		(testID.Protocol == TCP) {
+		if !IsAdmin() {
+			ui.printMsg("Warning: You are not running as administrator. For %s based %s",
+				protoToString(testID.Protocol), testToString(testID.Type))
+			ui.printMsg("test, running as administrator is required.\n")
+		}
+	}
+}
+
 func IsAdmin() bool {
-	_, err := os.Open("\\\\.\\PHYSICALDRIVE0")
+	c, err := os.Open("\\\\.\\PHYSICALDRIVE0")
 	if err != nil {
 		ui.printDbg("Process is not running as admin. Error: %v", err)
 		return false
 	}
+	c.Close()
 	return true
 }
+
+func SetTClass(fd uintptr, tos int) {
+	return
+}

server.go

@@ -6,7 +6,6 @@
 package main
 
 import (
-	"encoding/gob"
 	"fmt"
 	"io"
 	"net"
@@ -30,15 +29,15 @@ func finiServer() {
 
 func showAcceptedIPVersion() {
 	var ipVerString = "ipv4, ipv6"
-	if ipVer == ethrIPv4 {
+	if gIPVersion == ethrIPv4 {
 		ipVerString = "ipv4"
-	} else if ipVer == ethrIPv6 {
+	} else if gIPVersion == ethrIPv6 {
 		ipVerString = "ipv6"
 	}
 	ui.printMsg("Accepting IP version: %s", ipVerString)
 }
 
-func runServer(testParam EthrTestParam, serverParam ethrServerParam) {
+func runServer(serverParam ethrServerParam) {
 	defer stopStatsTimer()
 	initServer(serverParam.showUI)
 	startStatsTimer()
@@ -49,34 +48,27 @@ func runServer(testParam EthrTestParam, serverParam ethrServerParam) {
 	err := srvrRunTCPServer()
 	if err != nil {
 		finiServer()
-		fmt.Printf("Fatal error running TCP server: %v", err)
+		fmt.Printf("Fatal error running TCP server: %v\n", err)
 		os.Exit(1)
 	}
 }
 
-func handshakeWithClient(test *ethrTest, conn net.Conn) (testParam EthrTestParam, err error) {
+func handshakeWithClient(test *ethrTest, conn net.Conn) (testID EthrTestID, clientParam EthrClientParam, err error) {
-	// Check if there is any control message being sent to indicate type
+	ethrMsg := recvSessionMsg(conn)
-	// of test, the client is running.
-	dec := gob.NewDecoder(conn)
-	enc := gob.NewEncoder(conn)
-	ethrMsg := recvSessionMsg(dec)
 	if ethrMsg.Type != EthrSyn {
-		// For CPS & Connection Latency tests, this function will return here.
+		ui.printDbg("Failed to receive SYN message from client.")
 		err = os.ErrInvalid
 		return
 	}
-	testParam = ethrMsg.Syn.TestParam
+	testID = ethrMsg.Syn.TestID
-	delay := timeToNextTick()
+	clientParam = ethrMsg.Syn.ClientParam
-	ethrMsg = createAckMsg(gCert, delay)
+	ethrMsg = createAckMsg()
-	err = sendSessionMsg(enc, ethrMsg)
+	err = sendSessionMsg(conn, ethrMsg)
-	if err != nil {
-		ui.printErr("Send session message failed: %v", err)
-	}
 	return
 }
 
 func srvrRunTCPServer() error {
-	l, err := net.Listen(tcp(ipVer), hostAddr+":"+gEthrPortStr)
+	l, err := net.Listen(Tcp(), gLocalIP+":"+gEthrPortStr)
 	if err != nil {
 		return err
 	}
@@ -116,15 +108,18 @@ func srvrHandleNewTcpConn(conn net.Conn) {
 		ui.emitTestHdr()
 	}
 
-	// For CPS and ConnectionLatency tests, there is no deterministic way to know when
+	isCPSorPing := true
-	// the test starts from the client side and when it ends. This defer function ensures
+	// For CPS and Ping tests, there is no deterministic way to know when the test starts
-	// that test is not created/deleted repeatedly by doing a deferred deletion. If another
+	// from the client side and when it ends. This defer function ensures that test is not
-	// connection comes with-in 100ms, then another reference would be taken on existing
+	// created/deleted repeatedly by doing a deferred deletion. If another connection
-	// test object and it won't be deleted by safeDeleteTest call. This also ensures,
+	// comes with-in 2s, then another reference would be taken on existing test object
-	// test header is not printed repeatedly via emitTestHdr.
+	// and it won't be deleted by safeDeleteTest call. This also ensures, test header is
+	// not printed repeatedly via emitTestHdr.
 	// Note: Similar mechanism is used in UDP tests to handle test lifetime as well.
 	defer func() {
-		time.Sleep(100 * time.Millisecond)
+		if isCPSorPing {
+			time.Sleep(2 * time.Second)
+		}
 		safeDeleteTest(test)
 	}()
 
@@ -132,30 +127,35 @@ func srvrHandleNewTcpConn(conn net.Conn) {
 	// etc. and handle those cases as well.
 	atomic.AddUint64(&test.testResult.cps, 1)
 
-	testParam, err := handshakeWithClient(test, conn)
+	testID, clientParam, err := handshakeWithClient(test, conn)
 	if err != nil {
+		ui.printDbg("Failed in handshake with the client. Error: %v", err)
 		return
 	}
-
+	isCPSorPing = false
-	if testParam.TestID.Protocol == TCP {
+	if testID.Protocol == TCP {
-		if testParam.TestID.Type == Bandwidth {
+		if testID.Type == Bandwidth {
-			srvrRunTCPBandwidthTest(test, testParam, conn)
+			srvrRunTCPBandwidthTest(test, clientParam, conn)
-		} else if testParam.TestID.Type == Latency {
+		} else if testID.Type == Latency {
 			ui.emitLatencyHdr()
-			srvrRunTCPLatencyTest(test, testParam, conn)
+			srvrRunTCPLatencyTest(test, clientParam, conn)
 		}
 	}
 }
 
-func srvrRunTCPBandwidthTest(test *ethrTest, testParam EthrTestParam, conn net.Conn) {
+func srvrRunTCPBandwidthTest(test *ethrTest, clientParam EthrClientParam, conn net.Conn) {
-	size := testParam.BufferSize
+	size := clientParam.BufferSize
+	if clientParam.BwRate > 0 && uint64(size) > clientParam.BwRate {
+		size = uint32(clientParam.BwRate)
+	}
 	buff := make([]byte, size)
-	for i := uint32(0); i < testParam.BufferSize; i++ {
+	for i := uint32(0); i < size; i++ {
 		buff[i] = byte(i)
 	}
+	start, waitTime, sendRate := beginThrottle()
 	for {
 		var err error
-		if testParam.Reverse {
+		if clientParam.Reverse {
 			_, err = conn.Write(buff)
 		} else {
 			_, err = io.ReadFull(conn, buff)
@@ -164,13 +164,17 @@ func srvrRunTCPBandwidthTest(test *ethrTest, testParam EthrTestParam, conn net.C
 			ui.printDbg("Error sending/receiving data on a connection for bandwidth test: %v", err)
 			break
 		}
+		sendRate += uint64(size)
 		atomic.AddUint64(&test.testResult.bw, uint64(size))
+		if clientParam.BwRate > 0 && clientParam.Reverse && sendRate >= clientParam.BwRate {
+			start, waitTime, sendRate = enforceThrottle(start, waitTime)
+		}
 	}
 }
 
-func srvrRunTCPLatencyTest(test *ethrTest, testParam EthrTestParam, conn net.Conn) {
+func srvrRunTCPLatencyTest(test *ethrTest, clientParam EthrClientParam, conn net.Conn) {
-	bytes := make([]byte, testParam.BufferSize)
+	bytes := make([]byte, clientParam.BufferSize)
-	rttCount := testParam.RttCount
+	rttCount := clientParam.RttCount
 	latencyNumbers := make([]time.Duration, rttCount)
 	for {
 		_, err := io.ReadFull(conn, bytes)
@@ -222,18 +226,18 @@ func srvrRunTCPLatencyTest(test *ethrTest, testParam EthrTestParam, conn net.Con
 		p9999 := latencyNumbers[uint64(((float64(rttCountFixed)*99.99)/100)-1)]
 		ui.emitLatencyResults(
 			test.session.remoteIP,
-			protoToString(test.testParam.TestID.Protocol),
+			protoToString(test.testID.Protocol),
 			avg, min, max, p50, p90, p95, p99, p999, p9999)
 	}
 }
 
 func srvrRunUDPServer() error {
-	udpAddr, err := net.ResolveUDPAddr(udp(ipVer), hostAddr+":"+gEthrPortStr)
+	udpAddr, err := net.ResolveUDPAddr(Udp(), gLocalIP+":"+gEthrPortStr)
 	if err != nil {
 		ui.printDbg("Unable to resolve UDP address: %v", err)
 		return err
 	}
-	l, err := net.ListenUDP(udp(ipVer), udpAddr)
+	l, err := net.ListenUDP(Udp(), udpAddr)
 	if err != nil {
 		ui.printDbg("Error listening on %s for UDP pkt/s tests: %v", gEthrPortStr, err)
 		return err
@@ -255,7 +259,7 @@ func srvrRunUDPPacketHandler(conn *net.UDPConn) {
 	// address. We could use createOrGetTest but that takes a global lock.
 	tests := make(map[string]*ethrTest)
 	// For UDP, allocate buffer that can accomodate largest UDP datagram.
-	buffer := make([]byte, 64*1024)
+	readBuffer := make([]byte, 64*1024)
 	n, remoteIP, err := 0, new(net.UDPAddr), error(nil)
 
 	// This function handles UDP tests that came from clients that are no longer
@@ -264,10 +268,17 @@ func srvrRunUDPPacketHandler(conn *net.UDPConn) {
 	// has no reliable way to detect if client is active or not.
 	go func() {
 		for {
-			time.Sleep(200 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			for k, v := range tests {
 				ui.printDbg("Found Test from server: %v, time: %v", k, v.lastAccess)
-				if time.Since(v.lastAccess) > (100 * time.Millisecond) {
+				// At 200ms of no activity, mark the test in-active so stats stop
+				// printing.
+				if time.Since(v.lastAccess) > (200 * time.Millisecond) {
+					v.isDormant = true
+				}
+				// At 2s of no activity, delete the test by assuming that client
+				// has stopped.
+				if time.Since(v.lastAccess) > (2 * time.Second) {
 					ui.printDbg("Deleting UDP test from server: %v, lastAccess: %v", k, v.lastAccess)
 					safeDeleteTest(v)
 					delete(tests, k)
@@ -276,7 +287,7 @@ func srvrRunUDPPacketHandler(conn *net.UDPConn) {
 		}
 	}()
 	for err == nil {
-		n, remoteIP, err = conn.ReadFromUDP(buffer)
+		n, remoteIP, err = conn.ReadFromUDP(readBuffer)
 		if err != nil {
 			ui.printDbg("Error receiving data from UDP for bandwidth test: %v", err)
 			continue
@@ -295,6 +306,7 @@ func srvrRunUDPPacketHandler(conn *net.UDPConn) {
 			}
 		}
 		if test != nil {
+			test.isDormant = false
 			test.lastAccess = time.Now()
 			atomic.AddUint64(&test.testResult.pps, 1)
 			atomic.AddUint64(&test.testResult.bw, uint64(n))

serverui.go

@@ -29,8 +29,6 @@ var gAggregateTestResults = make(map[EthrProtocol]*ethrTestResultAggregate)
 func initServerUI(showUI bool) {
 	gAggregateTestResults[TCP] = &ethrTestResultAggregate{}
 	gAggregateTestResults[UDP] = &ethrTestResultAggregate{}
-	gAggregateTestResults[HTTP] = &ethrTestResultAggregate{}
-	gAggregateTestResults[HTTPS] = &ethrTestResultAggregate{}
 	gAggregateTestResults[ICMP] = &ethrTestResultAggregate{}
 	if !showUI || !initServerTui() {
 		initServerCli()
@@ -369,7 +367,7 @@ func (u *serverCli) printTestResults(s []string) {
 }
 
 func emitAggregateResults() {
-	var protoList = []EthrProtocol{TCP, UDP, HTTP, HTTPS, ICMP}
+	var protoList = []EthrProtocol{TCP, UDP, ICMP}
 	for _, proto := range protoList {
 		emitAggregate(proto)
 	}
@@ -397,7 +395,6 @@ func emitAggregate(proto EthrProtocol) {
 }
 
 func getTestResults(s *ethrSession, proto EthrProtocol, seconds uint64) []string {
-
 	var bwTestOn, cpsTestOn, ppsTestOn, latTestOn bool
 	var bw, cps, pps, latency uint64
 	aggTestResult, _ := gAggregateTestResults[proto]
@@ -431,6 +428,10 @@ func getTestResults(s *ethrSession, proto EthrProtocol, seconds uint64) []string
 				latTestOn = true
 			}
 		}
+
+		if test.isDormant && !((bwTestOn && bw != 0) || (cpsTestOn && cps != 0) || (ppsTestOn && pps != 0) || (latTestOn && latency != 0)) {
+			return []string{}
+		}
 	}
 
 	if bwTestOn || cpsTestOn || ppsTestOn || latTestOn {
@@ -454,60 +455,3 @@ func getTestResults(s *ethrSession, proto EthrProtocol, seconds uint64) []string
 
 	return []string{}
 }
-
-/*
-func getTestResults(s *ethrSession, proto EthrProtocol, seconds uint64) []string {
-	var bwTestOn, cpsTestOn, ppsTestOn, latTestOn bool
-	var bw, cps, pps, latency uint64
-	aggTestResult, _ := gAggregateTestResults[proto]
-	test, found := s.tests[EthrTestID{proto, Bandwidth}]
-	if found && test.isActive {
-		bwTestOn = true
-		bw = atomic.SwapUint64(&test.testResult.data, 0)
-		bw /= seconds
-		aggTestResult.bw += bw
-		aggTestResult.cbw++
-	}
-	test, found = s.tests[EthrTestID{proto, Cps}]
-	if found && test.isActive {
-		cpsTestOn = true
-		cps = atomic.SwapUint64(&test.testResult.data, 0)
-		cps /= seconds
-		aggTestResult.cps += cps
-		aggTestResult.ccps++
-	}
-	test, found = s.tests[EthrTestID{proto, Pps}]
-	if found && test.isActive {
-		ppsTestOn = true
-		pps = atomic.SwapUint64(&test.testResult.data, 0)
-		pps /= seconds
-		aggTestResult.pps += pps
-		aggTestResult.cpps++
-	}
-	test, found = s.tests[EthrTestID{proto, Latency}]
-	if found && test.isActive {
-		latTestOn = true
-		latency = atomic.LoadUint64(&test.testResult.data)
-	}
-	if bwTestOn || cpsTestOn || ppsTestOn || latTestOn {
-		var bwStr, cpsStr, ppsStr, latStr string
-		if bwTestOn {
-			bwStr = bytesToRate(bw)
-		}
-		if cpsTestOn {
-			cpsStr = cpsToString(cps)
-		}
-		if ppsTestOn {
-			ppsStr = ppsToString(pps)
-		}
-		if latTestOn {
-			latStr = durationToString(time.Duration(latency))
-		}
-		str := []string{s.remoteIP, protoToString(proto),
-			bwStr, cpsStr, ppsStr, latStr}
-		return str
-	}
-
-	return []string{}
-}
-*/

session.go

@@ -6,6 +6,7 @@
 package main
 
 import (
+	"bytes"
 	"container/list"
 	"encoding/gob"
 	"net"
@@ -15,185 +16,85 @@ import (
 	"time"
 )
 
-// EthrTestType represents the test type.
 type EthrTestType uint32
 
 const (
-	// All represents all tests - For now only applicable for servers
 	All EthrTestType = iota
-
-	// Bandwidth represents the bandwidth test.
 	Bandwidth
-
-	// Cps represents connections/s test.
 	Cps
-
-	// Pps represents packets/s test.
 	Pps
-
-	// Latency represents the latency test.
 	Latency
-
-	// Ping test.
 	Ping
-
-	// TraceRoute
 	TraceRoute
-
-	// MyTraceRoute
 	MyTraceRoute
 )
 
-// EthrProtocol represents the network protocol.
 type EthrProtocol uint32
 
 const (
-	// TCP represents the tcp protocol.
 	TCP EthrProtocol = iota
-
-	// UDP represents the udp protocol.
 	UDP
-
-	// HTTP represents using http protocol.
-	HTTP
-
-	// HTTPS represents using https protocol.
-	HTTPS
-
-	// ICMP represents the icmp protocol.
 	ICMP
 )
 
-// EthrTestID represents the test id.
+const (
-type EthrTestID struct {
+	ICMPv4 = 1  // ICMP for IPv4
-	// Protocol represents the protocol this test uses.
+	ICMPv6 = 58 // ICMP for IPv6
-	Protocol EthrProtocol
+)
 
-	// Type represents the test type this test uses.
+type EthrTestID struct {
-	Type EthrTestType
+	Protocol EthrProtocol
+	Type     EthrTestType
 }
 
-// EthrMsgType represents the message type.
 type EthrMsgType uint32
 
 const (
-	// EthrInv represents the Inv message.
 	EthrInv EthrMsgType = iota
-
-	// EthrSyn represents the Syn message.
 	EthrSyn
-
-	// EthrAck represents the Ack message.
 	EthrAck
-
-	// EthrFin represents the Fin message.
-	EthrFin
-
-	// EthrBgn represents the Bgn message.
-	EthrBgn
-
-	// EthrEnd represents the End message.
-	EthrEnd
 )
 
-// EthrMsgVer represents the message version.
 type EthrMsgVer uint32
 
-// EthrMsg represents the message entity.
 type EthrMsg struct {
-	// Version represents the message version.
 	Version EthrMsgVer
-
+	Type    EthrMsgType
-	// Type represents the message type.
+	Syn     *EthrMsgSyn
-	Type EthrMsgType
+	Ack     *EthrMsgAck
-
-	// Syn represents the Syn value.
-	Syn *EthrMsgSyn
-
-	// Ack represents the Ack value.
-	Ack *EthrMsgAck
-
-	// Fin represents the Fin value.
-	Fin *EthrMsgFin
-
-	// Bgn represents the Bgn value.
-	Bgn *EthrMsgBgn
-
-	// End represents the End value.
-	End *EthrMsgEnd
 }
 
-// EthrMsgSyn represents the Syn entity.
 type EthrMsgSyn struct {
-	// TestParam represents the test parameters.
+	TestID      EthrTestID
-	TestParam EthrTestParam
+	ClientParam EthrClientParam
 }
 
-// EthrMsgAck represents the Ack entity.
 type EthrMsgAck struct {
-	Cert        []byte
-	NapDuration time.Duration
-}
-
-// EthrMsgFin represents the Fin entity.
-type EthrMsgFin struct {
-	// Message represents the message body.
-	Message string
-}
-
-// EthrMsgBgn represents the Bgn entity.
-type EthrMsgBgn struct {
-	// UDPPort represents the udp port.
-	UDPPort string
-}
-
-// EthrMsgEnd represents the End entity.
-type EthrMsgEnd struct {
-	// Message represents the message body.
-	Message string
-}
-
-// EthrTestParam represents the parameters used for the test.
-type EthrTestParam struct {
-	// TestID represents the test id of this test.
-	TestID EthrTestID
-
-	// NumThreads represents how many threads are used for the test.
-	NumThreads uint32
-
-	// BufferSize represents the buffer size.
-	BufferSize uint32
-
-	// RttCount represents the rtt count.
-	RttCount uint32
-
-	// Reverse mode for bandwidth tests.
-	Reverse bool
 }
 
 type ethrTestResult struct {
-	bw       uint64
+	bw      uint64
-	cps      uint64
+	cps     uint64
-	pps      uint64
+	pps     uint64
-	latency  uint64
+	latency uint64
-	clatency uint64
+	// clatency uint64
 }
 
 type ethrTest struct {
-	isActive   bool
+	isActive    bool
-	session    *ethrSession
+	isDormant   bool
-	remoteAddr string
+	session     *ethrSession
-	remoteIP   string
+	remoteAddr  string
-	remotePort string
+	remoteIP    string
-	dialAddr   string
+	remotePort  string
-	ctrlConn   net.Conn
+	dialAddr    string
-	refCount   int32
+	refCount    int32
-	rcvdMsgs   chan *EthrMsg
+	testID      EthrTestID
-	testParam  EthrTestParam
+	clientParam EthrClientParam
-	testResult ethrTestResult
+	testResult  ethrTestResult
-	done       chan struct{}
+	done        chan struct{}
-	connList   *list.List
+	connList    *list.List
-	lastAccess time.Time
+	lastAccess  time.Time
 }
 
 type ethrIPVer uint32
@@ -204,18 +105,24 @@ const (
 	ethrIPv6
 )
 
-type ethrClientParam struct {
+type EthrClientParam struct {
-	duration    time.Duration
+	NumThreads  uint32
-	gap         time.Duration
+	BufferSize  uint32
-	warmupCount int
+	RttCount    uint32
+	Reverse     bool
+	Duration    time.Duration
+	Gap         time.Duration
+	WarmupCount uint32
+	BwRate      uint64
+	ToS         uint8
 }
 
 type ethrServerParam struct {
 	showUI bool
 }
 
-var ipVer ethrIPVer = ethrIPAny
+var gIPVersion ethrIPVer = ethrIPAny
-var xMode bool
+var gIsExternalClient bool
 
 type ethrConn struct {
 	bw      uint64
@@ -248,13 +155,13 @@ func deleteKey(key string) {
 	gSessionKeys = gSessionKeys[:i]
 }
 
-func newTest(remoteIP string, conn net.Conn, testParam EthrTestParam) (*ethrTest, error) {
+func newTest(remoteIP string, testID EthrTestID, clientParam EthrClientParam) (*ethrTest, error) {
 	gSessionLock.Lock()
 	defer gSessionLock.Unlock()
-	return newTestInternal(remoteIP, conn, testParam)
+	return newTestInternal(remoteIP, testID, clientParam)
 }
 
-func newTestInternal(remoteIP string, conn net.Conn, testParam EthrTestParam) (*ethrTest, error) {
+func newTestInternal(remoteIP string, testID EthrTestID, clientParam EthrClientParam) (*ethrTest, error) {
 	var session *ethrSession
 	session, found := gSessions[remoteIP]
 	if !found {
@@ -265,21 +172,21 @@ func newTestInternal(remoteIP string, conn net.Conn, testParam EthrTestParam) (*
 		gSessionKeys = append(gSessionKeys, remoteIP)
 	}
 
-	test, found := session.tests[testParam.TestID]
+	test, found := session.tests[testID]
 	if found {
 		return test, os.ErrExist
 	}
 	session.testCount++
 	test = &ethrTest{}
 	test.session = session
-	test.ctrlConn = conn
 	test.refCount = 0
-	test.rcvdMsgs = make(chan *EthrMsg)
+	test.testID = testID
-	test.testParam = testParam
+	test.clientParam = clientParam
 	test.done = make(chan struct{})
 	test.connList = list.New()
 	test.lastAccess = time.Now()
-	session.tests[testParam.TestID] = test
+	test.isDormant = true
+	session.tests[testID] = test
 
 	return test, nil
 }
@@ -292,7 +199,7 @@ func deleteTest(test *ethrTest) {
 
 func deleteTestInternal(test *ethrTest) {
 	session := test.session
-	testID := test.testParam.TestID
+	testID := test.testID
 	//
 	// TODO fix this, we need to decide where to close this, inside this
 	// function or by the caller. The reason we may need it to be done by
@@ -340,8 +247,8 @@ func createOrGetTest(remoteIP string, proto EthrProtocol, testType EthrTestType)
 	test = getTestInternal(remoteIP, proto, testType)
 	if test == nil {
 		isNew = true
-		testParam := EthrTestParam{TestID: EthrTestID{proto, testType}}
+		testID := EthrTestID{proto, testType}
-		test, _ = newTestInternal(remoteIP, nil, testParam)
+		test, _ = newTestInternal(remoteIP, testID, EthrClientParam{})
 		test.isActive = true
 	}
 	atomic.AddInt32(&test.refCount, 1)
@@ -395,35 +302,72 @@ func (test *ethrTest) connListDo(f func(*ethrConn)) {
 	}
 }
 
-func recvSessionMsg(dec *gob.Decoder) (ethrMsg *EthrMsg) {
+func createSynMsg(testID EthrTestID, clientParam EthrClientParam) (ethrMsg *EthrMsg) {
-	ethrMsg = &EthrMsg{}
+	ethrMsg = &EthrMsg{Version: 0, Type: EthrSyn}
-	err := dec.Decode(ethrMsg)
+	ethrMsg.Syn = &EthrMsgSyn{}
-	if err != nil {
+	ethrMsg.Syn.TestID = testID
-		ui.printDbg("Error receiving message on control channel: %v", err)
+	ethrMsg.Syn.ClientParam = clientParam
-		ethrMsg.Type = EthrInv
-	}
 	return
 }
 
-func sendSessionMsg(enc *gob.Encoder, ethrMsg *EthrMsg) error {
+func createAckMsg() (ethrMsg *EthrMsg) {
-	err := enc.Encode(ethrMsg)
+	ethrMsg = &EthrMsg{Version: 0, Type: EthrAck}
+	ethrMsg.Ack = &EthrMsgAck{}
+	return
+}
+
+func recvSessionMsg(conn net.Conn) (ethrMsg *EthrMsg) {
+	ethrMsg = &EthrMsg{}
+	ethrMsg.Type = EthrInv
+	// TODO: Assuming max ethr message size as 4096 sent over gob.
+	msgBytes := make([]byte, 4096)
+	n, err := conn.Read(msgBytes)
+	if err != nil {
+		ui.printDbg("Error receiving message on control channel. Error: %v", err)
+		return
+	}
+	ethrMsg = decodeMsg(msgBytes[:n])
+	return
+}
+
+func recvSessionMsgFromBuffer(msgBytes []byte) (ethrMsg *EthrMsg) {
+	ethrMsg = decodeMsg(msgBytes)
+	return
+}
+
+func sendSessionMsg(conn net.Conn, ethrMsg *EthrMsg) (err error) {
+	msgBytes, err := encodeMsg(ethrMsg)
+	if err != nil {
+		ui.printDbg("Error sending message on control channel. Message: %v, Error: %v", ethrMsg, err)
+		return
+	}
+	_, err = conn.Write(msgBytes)
 	if err != nil {
 		ui.printDbg("Error sending message on control channel. Message: %v, Error: %v", ethrMsg, err)
 	}
 	return err
 }
 
-func createAckMsg(cert []byte, d time.Duration) (ethrMsg *EthrMsg) {
+func decodeMsg(msgBytes []byte) (ethrMsg *EthrMsg) {
-	ethrMsg = &EthrMsg{Version: 0, Type: EthrAck}
+	ethrMsg = &EthrMsg{}
-	ethrMsg.Ack = &EthrMsgAck{}
+	buffer := bytes.NewBuffer(msgBytes)
-	ethrMsg.Ack.Cert = cert
+	decoder := gob.NewDecoder(buffer)
-	ethrMsg.Ack.NapDuration = d
+	err := decoder.Decode(ethrMsg)
+	if err != nil {
+		ui.printDbg("Failed to decode message using Gob: %v", err)
+		ethrMsg.Type = EthrInv
+	}
 	return
 }
 
-func createSynMsg(testParam EthrTestParam) (ethrMsg *EthrMsg) {
+func encodeMsg(ethrMsg *EthrMsg) (msgBytes []byte, err error) {
-	ethrMsg = &EthrMsg{Version: 0, Type: EthrSyn}
+	var writeBuffer bytes.Buffer
-	ethrMsg.Syn = &EthrMsgSyn{}
+	encoder := gob.NewEncoder(&writeBuffer)
-	ethrMsg.Syn.TestParam = testParam
+	err = encoder.Encode(ethrMsg)
+	if err != nil {
+		ui.printDbg("Failed to encode message using Gob: %v", err)
+		return
+	}
+	msgBytes = writeBuffer.Bytes()
 	return
 }

stats.go

@@ -140,7 +140,6 @@ func emitStats() {
 		seconds = 1
 	}
 	ui.emitTestResultBegin()
-	// ui.printMsg("Time: %v, Seconds: %v", lastStatsTime, seconds)
 	emitTestResults(uint64(seconds))
 	ui.emitTestResultEnd()
 	ui.emitStats(getNetworkStats())
@@ -154,8 +153,6 @@ func emitTestResults(s uint64) {
 		v := gSessions[k]
 		ui.emitTestResult(v, TCP, s)
 		ui.emitTestResult(v, UDP, s)
-		ui.emitTestResult(v, HTTP, s)
-		ui.emitTestResult(v, HTTPS, s)
 		ui.emitTestResult(v, ICMP, s)
 	}
 }

utils.go

@@ -9,7 +9,6 @@ import (
 	"fmt"
 	"net"
 	"os"
-	"regexp"
 	"strconv"
 	"strings"
 	"syscall"
@@ -18,37 +17,18 @@ import (
 	"unicode/utf8"
 )
 
-//
+var gLocalIP = ""
-// Regular expression to parse input for custom ports.
+var gEthrPort = uint16(8888)
-//
-var customPortRegex = regexp.MustCompile("(\\w+)=([0-9]+)")
-
-//
-// TODO: Use a better way to define ports. The core logic is:
-// Find a base port, such as 9999, and the Bandwidth is: base - 0,
-// Cps is base - 1, Pps is base - 2 and Latency is base - 3
-//
-const (
-	hostAddr = ""
-)
-
-var gEthrPort = 8888
 var gEthrPortStr = ""
+var gClientPort = uint16(0)
+var gTOS = uint8(0)
+var gTTL = uint8(0)
 
 const (
-	// UNO represents 1 unit.
+	UNO  = 1
-	UNO = 1
-
-	// KILO represents k.
 	KILO = 1000
-
-	// MEGA represents m.
 	MEGA = 1000 * 1000
-
-	// GIGA represents g.
 	GIGA = 1000 * 1000 * 1000
-
-	// TERA represents t.
 	TERA = 1000 * 1000 * 1000 * 1000
 )
 
@@ -142,6 +122,8 @@ func testToString(testType EthrTestType) string {
 		return "Ping"
 	case TraceRoute:
 		return "TraceRoute"
+	case MyTraceRoute:
+		return "MyTraceRoute"
 	default:
 		return "Invalid"
 	}
@@ -182,18 +164,14 @@ func protoToString(proto EthrProtocol) string {
 		return "TCP"
 	case UDP:
 		return "UDP"
-	case HTTP:
-		return "HTTP"
-	case HTTPS:
-		return "HTTPS"
 	case ICMP:
 		return "ICMP"
 	}
 	return ""
 }
 
-func tcp(ipVer ethrIPVer) string {
+func Tcp() string {
-	switch ipVer {
+	switch gIPVersion {
 	case ethrIPv4:
 		return "tcp4"
 	case ethrIPv6:
@@ -202,8 +180,8 @@ func tcp(ipVer ethrIPVer) string {
 	return "tcp"
 }
 
-func udp(ipVer ethrIPVer) string {
+func Udp() string {
-	switch ipVer {
+	switch gIPVersion {
 	case ethrIPv4:
 		return "udp4"
 	case ethrIPv6:
@@ -212,8 +190,8 @@ func udp(ipVer ethrIPVer) string {
 	return "udp"
 }
 
-func Icmp(ipVer ethrIPVer) string {
+func Icmp() string {
-	switch ipVer {
+	switch gIPVersion {
 	case ethrIPv6:
 		return "ip6:ipv6-icmp"
 	default:
@@ -221,11 +199,11 @@ func Icmp(ipVer ethrIPVer) string {
 	}
 }
 
-func IcmpProto(ipVer ethrIPVer) int {
+func IcmpProto() int {
-	if ipVer == ethrIPv6 {
+	if gIPVersion == ethrIPv6 {
-		return Icmpv6
+		return ICMPv6
 	}
-	return Icmpv4
+	return ICMPv4
 }
 
 func ethrUnused(vals ...interface{}) {
@@ -345,36 +323,80 @@ func SleepUntilNextWholeSecond() {
 	time.Sleep(time.Until(res))
 }
 
-func ethrDialForTraceRoute(network, address string, portNum uint16, ttl int) (conn net.Conn, err error) {
+func ethrSetTTL(fd uintptr, ttl int) {
-	port := fmt.Sprintf(":%v", portNum)
+	if ttl == 0 {
-	la, err := net.ResolveTCPAddr(network, port)
+		return
+	}
+	if gIPVersion == ethrIPv4 {
+		setSockOptInt(fd, syscall.IPPROTO_IP, syscall.IP_TTL, ttl)
+	} else {
+		setSockOptInt(fd, syscall.IPPROTO_IPV6, syscall.IPV6_UNICAST_HOPS, ttl)
+	}
+}
+
+func ethrSetTOS(fd uintptr, tos int) {
+	if tos == 0 {
+		return
+	}
+	if gIPVersion == ethrIPv4 {
+		setSockOptInt(fd, syscall.IPPROTO_IP, syscall.IP_TOS, tos)
+	} else {
+		SetTClass(fd, tos)
+	}
+}
+
+func ethrDial(p EthrProtocol, dialAddr string) (conn net.Conn, err error) {
+	return ethrDialEx(p, dialAddr, gLocalIP, gClientPort, int(gTTL), int(gTOS))
+}
+
+func ethrDialInc(p EthrProtocol, dialAddr string, inc uint16) (conn net.Conn, err error) {
+	if gClientPort != 0 {
+		return ethrDialEx(p, dialAddr, gLocalIP, gClientPort+inc, int(gTTL), int(gTOS))
+	} else {
+		return ethrDial(p, dialAddr)
+	}
+}
+
+func ethrDialAll(p EthrProtocol, dialAddr string) (conn net.Conn, err error) {
+	return ethrDialEx(p, dialAddr, gLocalIP, 0, int(gTTL), int(gTOS))
+}
+
+func ethrDialEx(p EthrProtocol, dialAddr, localIP string, localPortNum uint16, ttl int, tos int) (conn net.Conn, err error) {
+	localAddr := fmt.Sprintf("%v:%v", localIP, localPortNum)
+	var la net.Addr
+	network := Tcp()
+	if p == TCP {
+		la, err = net.ResolveTCPAddr(network, localAddr)
+	} else if p == UDP {
+		network = Udp()
+		la, err = net.ResolveUDPAddr(network, localAddr)
+	} else {
+		ui.printDbg("Only TCP or UDP are allowed in ethrDial")
+		err = os.ErrInvalid
+		return
+	}
 	if err != nil {
-		ui.printErr("Unable to resolve TCP address. Error: %v", err)
+		ui.printErr("Unable to resolve TCP or UDP address. Error: %v", err)
 		return
 	}
 	dialer := &net.Dialer{
 		Control: func(network, address string, c syscall.RawConn) error {
 			return c.Control(func(fd uintptr) {
-				if ipVer == ethrIPv4 {
+				ethrSetTTL(fd, ttl)
-					setSockOptInt(fd, syscall.IPPROTO_IP, syscall.IP_TTL, ttl)
+				ethrSetTOS(fd, tos)
-				} else {
-					setSockOptInt(fd, syscall.IPPROTO_IPV6, syscall.IPV6_UNICAST_HOPS, ttl)
-
-				}
 			})
 		},
 	}
 	dialer.LocalAddr = la
 	dialer.Timeout = time.Second
-	conn, err = dialer.Dial(network, address)
+	conn, err = dialer.Dial(network, dialAddr)
 	if err != nil {
-		ui.printDbg("ethrDialForTraceRoute Error: %v", err)
+		ui.printDbg("ethrTCPDial Error: %v", err)
 	} else {
 		tcpconn, ok := conn.(*net.TCPConn)
 		if ok {
 			tcpconn.SetLinger(0)
 		}
-		conn.Close()
 	}
 	return
 }
@@ -392,11 +414,11 @@ func ethrLookupIP(server string) (net.IPAddr, string, error) {
 
 	ips, err := net.LookupIP(server)
 	if err != nil {
-		ui.printErr("Failed to looup IP address for the server: %v. Error: %v", server, err)
+		ui.printErr("Failed to lookup IP address for the server: %v. Error: %v", server, err)
 		return ipAddr, ipStr, err
 	}
 	for _, ip := range ips {
-		if ipVer == ethrIPAny || (ipVer == ethrIPv4 && ip.To4() != nil) || (ipVer == ethrIPv6 && ip.To16() != nil) {
+		if gIPVersion == ethrIPAny || (gIPVersion == ethrIPv4 && ip.To4() != nil) || (gIPVersion == ethrIPv6 && ip.To16() != nil) {
 			ipAddr.IP = ip
 			ipStr = ip.String()
 			ui.printDbg("Resolved server: %v to IP address: %v\n", server, ip)
@@ -406,3 +428,24 @@ func ethrLookupIP(server string) (net.IPAddr, string, error) {
 	ui.printErr("Unable to resolve the given server: %v to an IP address.", server)
 	return ipAddr, ipStr, os.ErrNotExist
 }
+
+// This is a workaround to ensure we generate traffic at certain rate
+// and stats are printed correctly. We ensure that current interval lasts
+// 100ms after stats are printed, not perfect but workable.
+func beginThrottle() (start time.Time, waitTime time.Duration, sendRate uint64) {
+	start = time.Now()
+	waitTime = time.Until(lastStatsTime.Add(time.Second + 100*time.Millisecond))
+	sendRate = uint64(0)
+	return
+}
+
+func enforceThrottle(s time.Time, wt time.Duration) (start time.Time, waitTime time.Duration, sendRate uint64) {
+	timeTaken := time.Since(s)
+	if timeTaken < wt {
+		time.Sleep(wt - timeTaken)
+	}
+	start = time.Now()
+	waitTime = time.Until(lastStatsTime.Add(time.Second + 100*time.Millisecond))
+	sendRate = 0
+	return
+}