Support user-configured heartbeat intervals

There are two changes:
* The heartbeat interval period is now a user specified option,
where the default is 30s.
* A heartbeat is only transmitted if there was no outbound traffic.
This is an optimization that enables us to avoid transmitting a heartbeat
if a normal data packet would have served the same purpose.

Author: dsnet

logger.go

@@ -10,6 +10,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/dsnet/golib/unitconv"
@@ -20,7 +21,7 @@ type packetLogger struct {
 	last   time.Time // Last time we printed statistics
 	c      chan packetLog
 	m      map[packetLog]packetCount
-	rx, tx struct{ okay, drop packetCount }
+	rx, tx struct{ okay, drop packetCount } // Atomically updated
 }
 
 type packetLog struct {
@@ -74,19 +75,53 @@ func (pl *packetLogger) Log(b []byte, d direction, dropped bool) {
 	pl.c <- p
 }
 
+// Stats returns statistics on the total number and sizes of packets
+// transmitted or received.
+func (pl *packetLogger) Stats() (s struct {
+	Rx, Tx struct{ Okay, Drop struct{ Count, Sizes uint64 } }
+}) {
+	s.Tx.Okay.Count = atomic.LoadUint64(&pl.tx.okay.count)
+	s.Tx.Okay.Sizes = atomic.LoadUint64(&pl.tx.okay.sizes)
+	s.Rx.Okay.Count = atomic.LoadUint64(&pl.rx.okay.count)
+	s.Rx.Okay.Sizes = atomic.LoadUint64(&pl.rx.okay.sizes)
+	s.Tx.Drop.Count = atomic.LoadUint64(&pl.tx.drop.count)
+	s.Tx.Drop.Sizes = atomic.LoadUint64(&pl.tx.drop.sizes)
+	s.Rx.Drop.Count = atomic.LoadUint64(&pl.rx.drop.count)
+	s.Rx.Drop.Sizes = atomic.LoadUint64(&pl.rx.drop.sizes)
+	return
+}
+
 func (pl *packetLogger) monitor(ctx context.Context) {
 	t := time.NewTicker(30 * time.Second)
 	defer t.Stop()
 
 	for {
 		select {
 		case p := <-pl.c:
-			size := p.length
+			size := uint64(p.length)
+
+			// Update fine-granularity statistics.
 			p.length = 0
 			pc := pl.m[p]
 			pc.count++
-			pc.sizes += uint64(size)
+			pc.sizes += size
 			pl.m[p] = pc
+
+			// Update total packet statistics.
+			switch {
+			case !p.dropped && p.direction == outbound:
+				atomic.AddUint64(&pl.tx.okay.count, 1)
+				atomic.AddUint64(&pl.tx.okay.sizes, size)
+			case !p.dropped && p.direction == inbound:
+				atomic.AddUint64(&pl.rx.okay.count, 1)
+				atomic.AddUint64(&pl.rx.okay.sizes, size)
+			case p.dropped && p.direction == outbound:
+				atomic.AddUint64(&pl.tx.drop.count, 1)
+				atomic.AddUint64(&pl.tx.drop.sizes, size)
+			case p.dropped && p.direction == inbound:
+				atomic.AddUint64(&pl.rx.drop.count, 1)
+				atomic.AddUint64(&pl.rx.drop.sizes, size)
+			}
 		case <-t.C:
 			var count, sizes uint64
 			for _, v := range pl.m {
@@ -113,19 +148,6 @@ func (pl *packetLogger) print() {
 	stats := make([]string, 2) // First 2 lines for total stats
 	protoNames := map[int]string{icmp: "ICMP", udp: "UDP", tcp: "TCP"}
 	for k, v := range pl.m {
-		var pc *packetCount
-		switch {
-		case !k.dropped && k.direction == outbound:
-			pc = &pl.tx.okay
-		case !k.dropped && k.direction == inbound:
-			pc = &pl.rx.okay
-		case k.dropped && k.direction == outbound:
-			pc = &pl.tx.drop
-		case k.dropped && k.direction == inbound:
-			pc = &pl.rx.drop
-		}
-		pc.count += v.count
-		pc.sizes += v.sizes
 
 		proto := protoNames[k.ipProtocol]
 		if proto == "" {
@@ -142,10 +164,12 @@ func (pl *packetLogger) print() {
 		stats = append(stats, fmt.Sprintf("\tIPv%d/%s %s - %cx %d %spackets (%sB)",
 			k.ipVersion, proto, link, k.direction, v.count, drop, formatIEC(v.sizes)))
 	}
+
+	s := pl.Stats()
 	stats[0] = fmt.Sprintf("\tRx %d total packets (%sB), dropped %d total packets (%sB)",
-		pl.rx.okay.count, formatIEC(pl.rx.okay.sizes), pl.rx.drop.count, formatIEC(pl.rx.drop.sizes))
+		s.Rx.Okay.Count, formatIEC(s.Rx.Okay.Sizes), s.Rx.Drop.Count, formatIEC(s.Rx.Drop.Sizes))
 	stats[1] = fmt.Sprintf("\tTx %d total packets (%sB), dropped %d total packets (%sB)",
-		pl.tx.okay.count, formatIEC(pl.tx.okay.sizes), pl.tx.drop.count, formatIEC(pl.tx.drop.sizes))
+		s.Tx.Okay.Count, formatIEC(s.Tx.Okay.Sizes), s.Tx.Drop.Count, formatIEC(s.Tx.Drop.Sizes))
 	sort.Strings(stats[2:])
 	period := time.Now().Round(time.Second).Sub(pl.last)
 	pl.logger.Printf("Packet statistics (%v):\n%s", period, strings.Join(stats, "\n"))

main.go

@@ -52,6 +52,7 @@ import (
 	"path"
 	"runtime"
 	"syscall"
+	"time"
 
 	"github.com/dsnet/golib/jsonfmt"
 )
@@ -102,6 +103,15 @@ type TunnelConfig struct {
 	// The set of allowed ports must match on both the client and server.
 	AllowedPorts []uint16
 
+	// HeartbeatInterval is the amount of time in seconds without any
+	// outbound traffic to wait before the tunnel client will send a heartbeat
+	// message to the server. In the event that the client's address changed,
+	// this informs the server of the new client address.
+	//
+	// This field only applies to the client.
+	// The default value is 30.
+	HeartbeatInterval *uint
+
 	// PacketMagic is used to generate a sequence of bytes that is prepended to
 	// every TUN packet sent over UDP. Only inbound messages carrying the
 	// magic sequence will be accepted. This mechanism is used as a trivial way
@@ -148,6 +158,10 @@ func loadConfig(conf string) (tunn tunnel, logger *log.Logger, closer func() err
 	if config.TunnelAddress == config.TunnelPeerAddress {
 		logger.Fatal("TunnelAddress and TunnelPeerAddress must not conflict")
 	}
+	if config.HeartbeatInterval == nil {
+		config.HeartbeatInterval = new(uint)
+		*config.HeartbeatInterval = 30
+	}
 	host, _, _ := net.SplitHostPort(config.NetworkAddress)
 	serverMode := host == ""
 
@@ -195,6 +209,7 @@ func loadConfig(conf string) (tunn tunnel, logger *log.Logger, closer func() err
 		netAddr:       config.NetworkAddress,
 		ports:         config.AllowedPorts,
 		magic:         config.PacketMagic,
+		beatInterval:  time.Second * time.Duration(*config.HeartbeatInterval),
 		log:           logger,
 	}
 	return tunn, logger, closer

tunnel.go

@@ -41,6 +41,7 @@ type tunnel struct {
 	netAddr       string
 	ports         []uint16
 	magic         string
+	beatInterval  time.Duration
 
 	log logger
 
@@ -113,6 +114,12 @@ func (t tunnel) run(ctx context.Context) {
 	// TUN device and UDP socket have been set up. However, there is no good
 	// support for doing so currently: https://golang.org/issue/1435
 
+	if t.testReady != nil {
+		close(t.testReady)
+	}
+	pf := newPortFilter(t.ports)
+	pl := newPacketLogger(ctx, &wg, t.log)
+
 	// On the client, start some goroutines to accommodate for the dynamically
 	// changing environment that the client may be in.
 	magic := md5.Sum([]byte(t.magic))
@@ -141,27 +148,29 @@ func (t tunnel) run(ctx context.Context) {
 		// periodically ping the server to inform it of our new UDP address.
 		// Sending a packet with only the magic header is sufficient.
 		go func() {
-			ticker := time.NewTicker(30 * time.Second)
+			if t.beatInterval == 0 {
+				return
+			}
+			ticker := time.NewTicker(t.beatInterval)
 			defer ticker.Stop()
+			var prevTxn uint64
 			for range ticker.C {
-				if isDone(ctx) {
+				if isDone(ctx) { // Stop if done.
 					return
 				}
 				raddr := t.loadRemoteAddr()
-				if raddr == nil {
+				if raddr == nil { // Skip if no remote endpoint.
 					continue
 				}
-				sock.WriteToUDP(magic[:], raddr)
+				txn := pl.Stats().Tx.Okay.Count
+				if prevTxn == txn { // Only send if there is no outbound traffic
+					sock.WriteToUDP(magic[:], raddr)
+				}
+				prevTxn = txn
 			}
 		}()
 	}
 
-	if t.testReady != nil {
-		close(t.testReady)
-	}
-	pf := newPortFilter(t.ports)
-	pl := newPacketLogger(ctx, &wg, t.log)
-
 	// Handle outbound traffic.
 	wg.Add(1)
 	go func() {