port/builtin: use libnetwork UDP proxy

hack/test/docker-entrypoint.sh

@@ -96,5 +96,29 @@ function benchmark::iperf3_reverse::main(){
     benchmark::iperf3_reverse --net=slirp4netns --mtu=65520 --port-driver=builtin
     set +x
 }
+
+function benchmark::iperf3_reverse_udp(){
+    statedir=$(mktemp -d)
+    INFO "[benchmark:iperf3_reverse_udp] $@"
+    $ROOTLESSKIT --state-dir=$statedir $@ iperf3 -s > /dev/null &
+    rkpid=$!
+    # wait for socket to be available
+    sleep 3
+    rootlessctl="rootlessctl --socket=$statedir/api.sock"
+    portids=$($rootlessctl add-ports 127.0.0.1:5201:5201/tcp 127.0.0.1:5201:5201/udp)
+    $rootlessctl list-ports
+    sleep 3
+    $IPERF3C 127.0.0.1 -u -b 100G
+    $rootlessctl remove-ports $portids
+    kill $rkpid
+}
+
+function benchmark::iperf3_reverse_udp::main(){
+    set -x
+    benchmark::iperf3_reverse_udp --net=slirp4netns --mtu=65520 --port-driver=builtin
+    set +x
+}
+
 benchmark::iperf3::main
 benchmark::iperf3_reverse::main
+benchmark::iperf3_reverse_udp::main

pkg/port/builtin/builtin.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/rootless-containers/rootlesskit/pkg/msgutil"
 	"github.com/rootless-containers/rootlesskit/pkg/port"
+	"github.com/rootless-containers/rootlesskit/pkg/port/builtin/udpproxy"
 	"github.com/rootless-containers/rootlesskit/pkg/port/portutil"
 )
 
@@ -290,10 +291,37 @@ func startUDPRoutines(socketPath string, spec port.Spec, stopCh <-chan struct{},
 	if err != nil {
 		return err
 	}
+	udpp := &udpproxy.UDPProxy{
+		LogWriter: logWriter,
+		Listener:  c,
+		BackendDial: func() (*net.UDPConn, error) {
+			// get fd from the child as an SCM_RIGHTS cmsg
+			fd, err := connectToChildWithRetry(socketPath, spec, 10)
+			if err != nil {
+				return nil, err
+			}
+			f := os.NewFile(uintptr(fd), "")
+			defer f.Close()
+			fc, err := net.FileConn(f)
+			if err != nil {
+				return nil, err
+			}
+			uc, ok := fc.(*net.UDPConn)
+			if !ok {
+				return nil, errors.Errorf("file conn doesn't implement *net.UDPConn: %+v", fc)
+			}
+			return uc, nil
+		},
+	}
+	go udpp.Run()
 	go func() {
-		if err := copyConnToChild(c, socketPath, spec, stopCh); err != nil {
-			fmt.Fprintf(logWriter, "copyConnToChild: %v\n", err)
-			return
+		for {
+			select {
+			case <-stopCh:
+				// udpp.Close closes ln as well
+				udpp.Close()
+				return
+			}
 		}
 	}()
 	// no wait

pkg/port/builtin/udpproxy/udp_proxy.go

@@ -0,0 +1,150 @@
+// Package udpproxy is from https://raw.githubusercontent.com/docker/libnetwork/fec6476dfa21380bf8ee4d74048515d968c1ee63/cmd/proxy/udp_proxy.go
+package udpproxy
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+)
+
+const (
+	// UDPConnTrackTimeout is the timeout used for UDP connection tracking
+	UDPConnTrackTimeout = 90 * time.Second
+	// UDPBufSize is the buffer size for the UDP proxy
+	UDPBufSize = 65507
+)
+
+// A net.Addr where the IP is split into two fields so you can use it as a key
+// in a map:
+type connTrackKey struct {
+	IPHigh uint64
+	IPLow  uint64
+	Port   int
+}
+
+func newConnTrackKey(addr *net.UDPAddr) *connTrackKey {
+	if len(addr.IP) == net.IPv4len {
+		return &connTrackKey{
+			IPHigh: 0,
+			IPLow:  uint64(binary.BigEndian.Uint32(addr.IP)),
+			Port:   addr.Port,
+		}
+	}
+	return &connTrackKey{
+		IPHigh: binary.BigEndian.Uint64(addr.IP[:8]),
+		IPLow:  binary.BigEndian.Uint64(addr.IP[8:]),
+		Port:   addr.Port,
+	}
+}
+
+type connTrackMap map[connTrackKey]*net.UDPConn
+
+// UDPProxy is proxy for which handles UDP datagrams.
+// From libnetwork udp_proxy.go .
+type UDPProxy struct {
+	LogWriter      io.Writer
+	Listener       *net.UDPConn
+	BackendDial    func() (*net.UDPConn, error)
+	connTrackTable connTrackMap
+	connTrackLock  sync.Mutex
+}
+
+func (proxy *UDPProxy) replyLoop(proxyConn *net.UDPConn, clientAddr *net.UDPAddr, clientKey *connTrackKey) {
+	defer func() {
+		proxy.connTrackLock.Lock()
+		delete(proxy.connTrackTable, *clientKey)
+		proxy.connTrackLock.Unlock()
+		proxyConn.Close()
+	}()
+
+	readBuf := make([]byte, UDPBufSize)
+	for {
+		proxyConn.SetReadDeadline(time.Now().Add(UDPConnTrackTimeout))
+	again:
+		read, err := proxyConn.Read(readBuf)
+		if err != nil {
+			if err, ok := err.(*net.OpError); ok && err.Err == syscall.ECONNREFUSED {
+				// This will happen if the last write failed
+				// (e.g: nothing is actually listening on the
+				// proxied port on the container), ignore it
+				// and continue until UDPConnTrackTimeout
+				// expires:
+				goto again
+			}
+			return
+		}
+		for i := 0; i != read; {
+			written, err := proxy.Listener.WriteToUDP(readBuf[i:read], clientAddr)
+			if err != nil {
+				return
+			}
+			i += written
+		}
+	}
+}
+
+// Run starts forwarding the traffic using UDP.
+func (proxy *UDPProxy) Run() {
+	proxy.connTrackTable = make(connTrackMap)
+	readBuf := make([]byte, UDPBufSize)
+	for {
+		read, from, err := proxy.Listener.ReadFromUDP(readBuf)
+		if err != nil {
+			// NOTE: Apparently ReadFrom doesn't return
+			// ECONNREFUSED like Read do (see comment in
+			// UDPProxy.replyLoop)
+			if !isClosedError(err) {
+				fmt.Fprintf(proxy.LogWriter, "Stopping proxy on udp: %v\n", err)
+			}
+			break
+		}
+
+		fromKey := newConnTrackKey(from)
+		proxy.connTrackLock.Lock()
+		proxyConn, hit := proxy.connTrackTable[*fromKey]
+		if !hit {
+			proxyConn, err = proxy.BackendDial()
+			if err != nil {
+				fmt.Fprintf(proxy.LogWriter, "Can't proxy a datagram to udp: %v\n", err)
+				proxy.connTrackLock.Unlock()
+				continue
+			}
+			proxy.connTrackTable[*fromKey] = proxyConn
+			go proxy.replyLoop(proxyConn, from, fromKey)
+		}
+		proxy.connTrackLock.Unlock()
+		for i := 0; i != read; {
+			written, err := proxyConn.Write(readBuf[i:read])
+			if err != nil {
+				fmt.Fprintf(proxy.LogWriter, "Can't proxy a datagram to udp: %v\n", err)
+				break
+			}
+			i += written
+		}
+	}
+}
+
+// Close stops forwarding the traffic.
+func (proxy *UDPProxy) Close() {
+	proxy.Listener.Close()
+	proxy.connTrackLock.Lock()
+	defer proxy.connTrackLock.Unlock()
+	for _, conn := range proxy.connTrackTable {
+		conn.Close()
+	}
+}
+
+func isClosedError(err error) bool {
+	/* This comparison is ugly, but unfortunately, net.go doesn't export errClosing.
+	 * See:
+	 * http://golang.org/src/pkg/net/net.go
+	 * https://code.google.com/p/go/issues/detail?id=4337
+	 * https://groups.google.com/forum/#!msg/golang-nuts/0_aaCvBmOcM/SptmDyX1XJMJ
+	 */
+	return strings.HasSuffix(err.Error(), "use of closed network connection")
+}