Repository: scionproto/scion
Pull request #3634 (2020-01-23 07:58:29)

godispatcher: Split out a library that could be linked into apps


<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/scionproto/scion/3634)
<!-- Reviewable:end -->



Patch:
diff --git a/go/godispatcher/dispatcher/BUILD.bazel b/go/godispatcher/dispatcher/BUILD.bazel
new file mode 100644
index 000000000..14059f746
--- /dev/null
+++ b/go/godispatcher/dispatcher/BUILD.bazel
@@ -0,0 +1,47 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "dispatcher.go",
+        "overlay.go",
+        "scmp.go",
+        "table.go",
+    ],
+    importpath = "github.com/scionproto/scion/go/godispatcher/dispatcher",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//go/godispatcher/internal/metrics:go_default_library",
+        "//go/godispatcher/internal/registration:go_default_library",
+        "//go/godispatcher/internal/respool:go_default_library",
+        "//go/lib/addr:go_default_library",
+        "//go/lib/common:go_default_library",
+        "//go/lib/hpkt:go_default_library",
+        "//go/lib/l4:go_default_library",
+        "//go/lib/layers:go_default_library",
+        "//go/lib/log:go_default_library",
+        "//go/lib/overlay/conn:go_default_library",
+        "//go/lib/ringbuf:go_default_library",
+        "//go/lib/scmp:go_default_library",
+        "//go/lib/serrors:go_default_library",
+        "//go/lib/spkt:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["overlay_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//go/lib/addr:go_default_library",
+        "//go/lib/common:go_default_library",
+        "//go/lib/l4:go_default_library",
+        "//go/lib/l4/mock_l4:go_default_library",
+        "//go/lib/scmp:go_default_library",
+        "//go/lib/spkt:go_default_library",
+        "//go/lib/xtest:go_default_library",
+        "@com_github_golang_mock//gomock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+    ],
+)
diff --git a/go/godispatcher/dispatcher/dispatcher.go b/go/godispatcher/dispatcher/dispatcher.go
new file mode 100644
index 000000000..b16ebfee2
--- /dev/null
+++ b/go/godispatcher/dispatcher/dispatcher.go
@@ -0,0 +1,328 @@
+package dispatcher
+
+import (
+	"context"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/scionproto/scion/go/godispatcher/internal/metrics"
+	"github.com/scionproto/scion/go/godispatcher/internal/registration"
+	"github.com/scionproto/scion/go/godispatcher/internal/respool"
+	"github.com/scionproto/scion/go/lib/addr"
+	"github.com/scionproto/scion/go/lib/common"
+	"github.com/scionproto/scion/go/lib/hpkt"
+	"github.com/scionproto/scion/go/lib/log"
+	"github.com/scionproto/scion/go/lib/overlay/conn"
+	"github.com/scionproto/scion/go/lib/ringbuf"
+	"github.com/scionproto/scion/go/lib/scmp"
+	"github.com/scionproto/scion/go/lib/serrors"
+	"github.com/scionproto/scion/go/lib/spkt"
+)
+
+// OverflowLoggingInterval is the minimum amount of time that needs to
+// pass before another overflow logging message is printed (if needed).
+const OverflowLoggingInterval = 10 * time.Second
+
+// ReceiveBufferSize is the size of receive buffers used by the dispatcher.
+const ReceiveBufferSize = 1 << 20
+
+// Server is the main object allowing to create new SCION connections.
+type Server struct {
+	// routingTable is used to register new connections.
+	routingTable *IATable
+	ipv4Conn     net.PacketConn
+	ipv6Conn     net.PacketConn
+}
+
+// NewServer creates new instance of Server. Internally, it opens the dispatcher ports
+// for both IPv4 and IPv6. Returns error if the ports can't be opened.
+func NewServer(address string) (*Server, error) {
+	metaLogger := &throttledMetaLogger{
+		Logger:      log.Root(),
+		MinInterval: OverflowLoggingInterval,
+	}
+	ipv4Conn, err := openConn("udp4", address, metaLogger)
+	if err != nil {
+		return nil, err
+	}
+	ipv6Conn, err := openConn("udp6", address, metaLogger)
+	if err != nil {
+		ipv4Conn.Close()
+		return nil, err
+	}
+
+	return &Server{
+		routingTable: NewIATable(1024, 65535),
+		ipv4Conn:     ipv4Conn,
+		ipv6Conn:     ipv6Conn,
+	}, nil
+}
+
+// Serve starts reading packets from network and dispatching them to different connections.
+// The function blocks and returns if there's an error or when Close has been called.
+func (as *Server) Serve() error {
+	errChan := make(chan error)
+	go func() {
+		defer log.LogPanicAndExit()
+		netToRingDataplane := &NetToRingDataplane{
+			OverlayConn:  as.ipv4Conn,
+			RoutingTable: as.routingTable,
+		}
+		errChan <- netToRingDataplane.Run()
+	}()
+	go func() {
+		defer log.LogPanicAndExit()
+		netToRingDataplane := &NetToRingDataplane{
+			OverlayConn:  as.ipv6Conn,
+			RoutingTable: as.routingTable,
+		}
+		errChan <- netToRingDataplane.Run()
+	}()
+	return <-errChan
+}
+
+// Register creates a new connection.
+func (as *Server) Register(ctx context.Context, ia addr.IA, address *net.UDPAddr,
+	svc addr.HostSVC) (net.PacketConn, uint16, error) {
+
+	tableEntry := newTableEntry()
+	ref, err := as.routingTable.Register(ia, address, nil, svc, tableEntry)
+	if err != nil {
+		return nil, 0, err
+	}
+	conn := &Conn{
+		ipv4Conn:     as.ipv4Conn,
+		ipv6Conn:     as.ipv6Conn,
+		ring:         tableEntry.appIngressRing,
+		regReference: ref,
+	}
+	return conn, uint16(ref.UDPAddr().Port), nil
+}
+
+func (as *Server) Close() {
+	as.ipv4Conn.Close()
+	as.ipv6Conn.Close()
+}
+
+// Conn represents a connection bound to a specific SCION port/SVC.
+type Conn struct {
+	// ipv4Conn is used to send IPv4 packets.
+	ipv4Conn net.PacketConn
+	// ipv6Conn is used to send IPv6 packets.
+	ipv6Conn net.PacketConn
+	// ring is used to retrieve incoming packets.
+	ring *ringbuf.Ring
+	// regReference is the reference to the registration in the routing table.
+	regReference registration.RegReference
+}
+
+func (ac *Conn) WriteTo(p []byte, addr net.Addr) (int, error) {
+	var info spkt.ScnPkt
+	hpkt.ParseScnPkt(&info, p)
+	if err := registerIfSCMPRequest(ac.regReference, &info); err != nil {
+		log.Warn("SCMP Request ID error, packet still sent", "err", err)
+	}
+	isIPv6 := addr.(*net.UDPAddr).IP.To4() == nil
+	if isIPv6 {
+		return ac.ipv6Conn.WriteTo(p, addr)
+	} else {
+		return ac.ipv4Conn.WriteTo(p, addr)
+	}
+}
+
+// Write is optimized for the use by ConnHandler (avoids reparsing the packet).
+func (ac *Conn) Write(pkt *respool.Packet) (int, error) {
+	if err := registerIfSCMPRequest(ac.regReference, &pkt.Info); err != nil {
+		log.Warn("SCMP Request ID error, packet still sent", "err", err)
+	}
+	isIPv6 := pkt.OverlayRemote.IP.To4() == nil
+	if isIPv6 {
+		return pkt.SendOnConn(ac.ipv6Conn, pkt.OverlayRemote)
+	} else {
+		return pkt.SendOnConn(ac.ipv4Conn, pkt.OverlayRemote)
+	}
+}
+
+func (ac *Conn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
+	pkt := ac.Read()
+	if pkt == nil {
+		return 0, nil, serrors.New("Connection closed")
+	}
+	n = pkt.CopyTo(p)
+	pkt.Free()
+	return
+}
+
+// Read is optimized for the use by ConnHandler (avoids one copy).
+func (ac *Conn) Read() *respool.Packet {
+	entries := make(ringbuf.EntryList, 1)
+	n, _ := ac.ring.Read(entries, true)
+	if n < 0 {
+		// Ring was closed because app shut down its data socket.
+		return nil
+	}
+	pkt := entries[0].(*respool.Packet)
+	return pkt
+}
+
+func (ac *Conn) Close() error {
+	ac.regReference.Free()
+	ac.ring.Close()
+	return nil
+}
+
+func (ac *Conn) LocalAddr() net.Addr {
+	return ac.regReference.UDPAddr()
+}
+
+func (ac *Conn) SVCAddr() addr.HostSVC {
+	return ac.regReference.SVCAddr()
+}
+
+func (ac *Conn) SetDeadline(t time.Time) error {
+	panic("not implemented")
+}
+
+func (ac *Conn) SetReadDeadline(t time.Time) error {
+	panic("not implemented")
+}
+
+func (ac *Conn) SetWriteDeadline(t time.Time) error {
+	panic("not implemented")
+}
+
+// openConn opens an overlay socket that tracks additional socket information
+// such as packets dropped due to buffer full.
+//
+// Note that Go-style dual-stacked IPv4/IPv6 connections are not supported. If
+// network is udp, it will be treated as udp4.
+func openConn(network, address string, p SocketMetaHandler) (net.PacketConn, error) {
+	// We cannot allow the Go standard library to open both types of sockets
+	// because the socket options are specific to only one socket type, so we
+	// degrade udp to only udp4.
+	if network == "udp" {
+		network = "udp4"
+	}
+	listeningAddress, err := net.ResolveUDPAddr(network, address)
+	if err != nil {
+		return nil, common.NewBasicError("unable to construct UDP addr", err)
+	}
+	if network == "udp4" && listeningAddress.IP == nil {
+		listeningAddress.IP = net.IPv4zero
+	}
+	if network == "udp6" && listeningAddress.IP == nil {
+		listeningAddress.IP = net.IPv6zero
+	}
+
+	c, err := conn.New(listeningAddress, nil, &conn.Config{ReceiveBufferSize: ReceiveBufferSize})
+	if err != nil {
+		return nil, common.NewBasicError("unable to open conn", err)
+	}
+
+	return &overlayConnWrapper{Conn: c, Handler: p}, nil
+}
+
+// registerIfSCMPRequest registers the ID of the SCMP Request, if it is an
+// SCMP::General::EchoRequest, SCMP::General::TraceRouteRequest or SCMP::General::RecordPathRequest
+// packet. It also increments SCMP-related metrics.
+func registerIfSCMPRequest(ref registration.RegReference, packet *spkt.ScnPkt) error {
+	if scmpHdr, ok := packet.L4.(*scmp.Hdr); ok {
+		metrics.M.SCMPWritePkts(
+			metrics.SCMP{
+				Class: scmpHdr.Class.String(),
+				Type:  scmpHdr.Type.Name(scmpHdr.Class),
+			},
+		).Inc()
+		if !isSCMPGeneralRequest(scmpHdr) {
+			return nil
+		}
+		if id := getSCMPGeneralID(packet); id != 0 {
+			return ref.RegisterID(id)
+		}
+	}
+	return nil
+}
+
+// overlayConnWrapper wraps a specialized overlay conn into a net.PacketConn
+// implementation. Only *net.UDPAddr addressing is supported.
+type overlayConnWrapper struct {
+	// Conn is the wrapped overlay connection object.
+	conn.Conn
+	// Handler is used to customize how the connection treats socket
+	// metadata.
+	Handler SocketMetaHandler
+}
+
+func (o *overlayConnWrapper) ReadFrom(p []byte) (int, net.Addr, error) {
+	n, meta, err := o.Conn.Read(common.RawBytes(p))
+	if meta == nil {
+		return n, nil, err
+	}
+	o.Handler.Handle(meta)
+	return n, meta.Src, err
+}
+
+func (o *overlayConnWrapper) WriteTo(p []byte, a net.Addr) (int, error) {
+	udpAddr, ok := a.(*net.UDPAddr)
+	if !ok {
+		return 0, common.NewBasicError("address is not UDP", nil, "addr", a)
+	}
+	return o.Conn.WriteTo(common.RawBytes(p), udpAddr)
+}
+
+func (o *overlayConnWrapper) Close() error {
+	return o.Conn.Close()
+}
+
+func (o *overlayConnWrapper) LocalAddr() net.Addr {
+	return o.Conn.LocalAddr()
+}
+
+func (o *overlayConnWrapper) SetDeadline(t time.Time) error {
+	return o.Conn.SetDeadline(t)
+}
+
+func (o *overlayConnWrapper) SetReadDeadline(t time.Time) error {
+	return o.Conn.SetReadDeadline(t)
+}
+
+func (o *overlayConnWrapper) SetWriteDeadline(t time.Time) error {
+	return o.Conn.SetWriteDeadline(t)
+}
+
+// SocketMetaHandler processes OS socket metadata during reads.
+type SocketMetaHandler interface {
+	Handle(*conn.ReadMeta)
+}
+
+// throttledMetaLogger logs packets dropped due to full receive buffers,
+// with a configurable threshold on how often logging messages are printed.
+type throttledMetaLogger struct {
+	// Logger is used to print the logging messages.
+	Logger log.Logger
+	// MinInterval is the minimum duration of time that has passed since the
+	MinInterval time.Duration
+
+	mu sync.Mutex
+	// lastPrintTimestamp is the time when the previous logging message was
+	// printed.
+	lastPrintTimestamp time.Time
+	// lastPrintValue is the overflow value printed in the last logging message.
+	lastPrintValue uint32
+}
+
+func (p *throttledMetaLogger) Handle(m *conn.ReadMeta) {
+	p.mu.Lock()
+	if m.RcvOvfl != p.lastPrintValue && time.Since(p.lastPrintTimestamp) > p.MinInterval {
+		if m.RcvOvfl > p.lastPrintValue {
+			metrics.M.NetReadOverflows().Add(float64(m.RcvOvfl - p.lastPrintValue))
+		} else {
+			metrics.M.NetReadOverflows().Add(float64(m.RcvOvfl))
+		}
+		p.Logger.Debug("Detected socket overflow", "total_cnt", m.RcvOvfl)
+		p.lastPrintTimestamp = time.Now()
+		p.lastPrintValue = m.RcvOvfl
+	}
+	p.mu.Unlock()
+}
diff --git a/go/godispatcher/network/overlay.go b/go/godispatcher/dispatcher/overlay.go
similarity index 99%
rename from go/godispatcher/network/overlay.go
rename to go/godispatcher/dispatcher/overlay.go
index 14ed6e233..245d29418 100644
--- a/go/godispatcher/network/overlay.go
+++ b/go/godispatcher/dispatcher/overlay.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package network
+package dispatcher
 
 import (
 	"net"
diff --git a/go/godispatcher/network/scmp.go b/go/godispatcher/dispatcher/scmp.go
similarity index 99%
rename from go/godispatcher/network/scmp.go
rename to go/godispatcher/dispatcher/scmp.go
index e6c124869..4652b2d36 100644
--- a/go/godispatcher/network/scmp.go
+++ b/go/godispatcher/dispatcher/scmp.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package network
+package dispatcher
 
 import (
 	"github.com/scionproto/scion/go/lib/common"
diff --git a/go/godispatcher/network/table.go b/go/godispatcher/dispatcher/table.go
similarity index 94%
rename from go/godispatcher/network/table.go
rename to go/godispatcher/dispatcher/table.go
index f6c956fa5..01b2bebd4 100644
--- a/go/godispatcher/network/table.go
+++ b/go/godispatcher/dispatcher/table.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package network
+package dispatcher
 
 import (
 	"net"
@@ -34,13 +34,6 @@ func newTableEntry() *TableEntry {
 	}
 }
 
-func getBindIP(address *net.UDPAddr) net.IP {
-	if address == nil {
-		return nil
-	}
-	return address.IP
-}
-
 // IATable is a type-safe convenience wrapper around a generic routing table.
 type IATable struct {
 	registration.IATable
diff --git a/go/godispatcher/internal/respool/packet.go b/go/godispatcher/internal/respool/packet.go
index c8ea7c657..60f4788ba 100644
--- a/go/godispatcher/internal/respool/packet.go
+++ b/go/godispatcher/internal/respool/packet.go
@@ -80,6 +80,14 @@ func (pkt *Packet) Dup() {
 	pkt.mtx.Unlock()
 }
 
+// CopyTo copies the buffer into the provided bytearray. Returns number of bytes copied.
+func (pkt *Packet) CopyTo(p []byte) int {
+	n := len(pkt.buffer)
+	p = p[:n]
+	copy(p, pkt.buffer)
+	return n
+}
+
 // Free releases a reference to the packet. Free is safe to use from concurrent
 // goroutines.
 func (pkt *Packet) Free() {
diff --git a/go/godispatcher/main.go b/go/godispatcher/main.go
index 709c2eedf..4dfdbd9a2 100644
--- a/go/godispatcher/main.go
+++ b/go/godispatcher/main.go
@@ -141,7 +141,6 @@ func RunDispatcher(deleteSocketFlag bool, applicationSocket string, socketFileMo
 		}
 	}
 	dispatcher := &network.Dispatcher{
-		RoutingTable:      network.NewIATable(1024, 65535),
 		OverlaySocket:     fmt.Sprintf(":%d", overlayPort),
 		ApplicationSocket: applicationSocket,
 		SocketFileMode:    socketFileMode,
diff --git a/go/godispatcher/network/BUILD.bazel b/go/godispatcher/network/BUILD.bazel
index 71eb50414..36208c5a8 100644
--- a/go/godispatcher/network/BUILD.bazel
+++ b/go/godispatcher/network/BUILD.bazel
@@ -1,49 +1,20 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "go_default_library",
     srcs = [
         "app_socket.go",
         "dispatcher.go",
-        "overlay.go",
-        "scmp.go",
-        "table.go",
     ],
     importpath = "github.com/scionproto/scion/go/godispatcher/network",
     visibility = ["//visibility:public"],
     deps = [
+        "//go/godispatcher/dispatcher:go_default_library",
         "//go/godispatcher/internal/metrics:go_default_library",
-        "//go/godispatcher/internal/registration:go_default_library",
         "//go/godispatcher/internal/respool:go_default_library",
         "//go/lib/addr:go_default_library",
         "//go/lib/common:go_default_library",
-        "//go/lib/hpkt:go_default_library",
-        "//go/lib/l4:go_default_library",
-        "//go/lib/layers:go_default_library",
         "//go/lib/log:go_default_library",
-        "//go/lib/overlay/conn:go_default_library",
-        "//go/lib/ringbuf:go_default_library",
-        "//go/lib/scmp:go_default_library",
-        "//go/lib/serrors:go_default_library",
         "//go/lib/sock/reliable:go_default_library",
-        "//go/lib/spkt:go_default_library",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["overlay_test.go"],
-    embed = [":go_default_library"],
-    deps = [
-        "//go/lib/addr:go_default_library",
-        "//go/lib/common:go_default_library",
-        "//go/lib/l4:go_default_library",
-        "//go/lib/l4/mock_l4:go_default_library",
-        "//go/lib/scmp:go_default_library",
-        "//go/lib/spkt:go_default_library",
-        "//go/lib/xtest:go_default_library",
-        "@com_github_golang_mock//gomock:go_default_library",
-        "@com_github_stretchr_testify//assert:go_default_library",
-        "@com_github_stretchr_testify//require:go_default_library",
     ],
 )
diff --git a/go/godispatcher/network/app_socket.go b/go/godispatcher/network/app_socket.go
index 519ac4cdd..dee80a837 100644
--- a/go/godispatcher/network/app_socket.go
+++ b/go/godispatcher/network/app_socket.go
@@ -19,23 +19,20 @@ import (
 	"io"
 	"net"
 
+	"github.com/scionproto/scion/go/godispatcher/dispatcher"
 	"github.com/scionproto/scion/go/godispatcher/internal/metrics"
-	"github.com/scionproto/scion/go/godispatcher/internal/registration"
 	"github.com/scionproto/scion/go/godispatcher/internal/respool"
 	"github.com/scionproto/scion/go/lib/addr"
 	"github.com/scionproto/scion/go/lib/common"
 	"github.com/scionproto/scion/go/lib/log"
-	"github.com/scionproto/scion/go/lib/ringbuf"
-	"github.com/scionproto/scion/go/lib/scmp"
 	"github.com/scionproto/scion/go/lib/sock/reliable"
-	"github.com/scionproto/scion/go/lib/spkt"
 )
 
 // AppSocketServer accepts new connections coming from SCION apps, and
 // hands them off to the registration + dataplane handler.
 type AppSocketServer struct {
-	Listener    *reliable.Listener
-	ConnManager *AppConnManager
+	Listener   *reliable.Listener
+	DispServer *dispatcher.Server
 }
 
 func (s *AppSocketServer) Serve() error {
@@ -45,114 +42,81 @@ func (s *AppSocketServer) Serve() error {
 			return err
 		}
 		pconn := conn.(net.PacketConn)
-		s.ConnManager.Handle(pconn)
+		s.Handle(pconn)
 	}
 }
 
-// AppConnManager handles new connections coming from SCION applications.
-type AppConnManager struct {
-	RoutingTable *IATable
-	// IPv4OverlayConn is the network connection to which IPv4 egress traffic
-	// is sent.
-	IPv4OverlayConn net.PacketConn
-	// IPv6OverlayConn is the network connection to which IPv6 egress traffic
-	// is sent.
-	IPv6OverlayConn net.PacketConn
-}
-
 // Handle passes conn off to a per-connection state handler.
-func (h *AppConnManager) Handle(conn net.PacketConn) {
+func (h *AppSocketServer) Handle(conn net.PacketConn) {
 	ch := &AppConnHandler{
-		Conn:            conn,
-		RoutingTable:    h.RoutingTable,
-		IPv4OverlayConn: h.IPv4OverlayConn,
-		IPv6OverlayConn: h.IPv6OverlayConn,
-		Logger:          log.Root().New("clientID", fmt.Sprintf("%p", conn)),
+		Conn:   conn,
+		Logger: log.Root().New("clientID", fmt.Sprintf("%p", conn)),
 	}
 	go func() {
 		defer log.LogPanicAndExit()
-		ch.Handle()
+		ch.Handle(h.DispServer)
 	}()
 }
 
 // AppConnHandler handles a single SCION application connection.
 type AppConnHandler struct {
-	RoutingTable *IATable
 	// Conn is the local socket to which the application is connected.
-	Conn net.PacketConn
-	// IPv4OverlayConn is the network connection to which egress IPv4 traffic
-	// is sent.
-	IPv4OverlayConn net.PacketConn
-	// IPv6OverlayConn is the network connection to which egress IPv6 traffic
-	// is sent.
-	IPv6OverlayConn net.PacketConn
-	Logger          log.Logger
+	Conn     net.PacketConn
+	DispConn *dispatcher.Conn
+	Logger   log.Logger
 }
 
-func (h *AppConnHandler) Handle() {
+func (h *AppConnHandler) Handle(appServer *dispatcher.Server) {
 	h.Logger.Info("Accepted new client")
 	defer h.Logger.Info("Closed client socket")
 	defer h.Conn.Close()
 
-	ref, tableEntry, useIPv6, err := h.doRegExchange()
+	dispConn, err := h.doRegExchange(appServer)
 	if err != nil {
 		metrics.M.AppConnErrors().Inc()
 		h.Logger.Warn("registration error", "err", err)
 		return
 	}
-	defer ref.Free()
-	metrics.M.OpenSockets(metrics.SVC{Type: ref.SVCAddr().String()}).Inc()
-	defer metrics.M.OpenSockets(metrics.SVC{Type: ref.SVCAddr().String()}).Dec()
+	h.DispConn = dispConn.(*dispatcher.Conn)
+	defer h.DispConn.Close()
+	svc := h.DispConn.SVCAddr().String()
+	metrics.M.OpenSockets(metrics.SVC{Type: svc}).Inc()
+	defer metrics.M.OpenSockets(metrics.SVC{Type: svc}).Dec()
 
-	defer tableEntry.appIngressRing.Close()
 	go func() {
 		defer log.LogPanicAndExit()
-		h.RunRingToAppDataplane(tableEntry.appIngressRing)
+		h.RunRingToAppDataplane()
 	}()
 
-	conn := h.IPv4OverlayConn
-	if useIPv6 {
-		conn = h.IPv6OverlayConn
-	}
-	h.RunAppToNetDataplane(ref, conn)
+	h.RunAppToNetDataplane()
 }
 
 // doRegExchange manages an application's registration request, and returns a
 // reference to registered data that should be freed at the end of the
-// registration, information about allocated ring buffers, a boolean specifying
-// whether to use IPv6 egress instead of IPv4, and whether an error occurred.
-func (h *AppConnHandler) doRegExchange() (registration.RegReference, *TableEntry, bool, error) {
+// registration, information about allocated ring buffers and whether an error occurred.
+func (h *AppConnHandler) doRegExchange(appServer *dispatcher.Server) (net.PacketConn, error) {
+
 	b := respool.GetBuffer()
 	defer respool.PutBuffer(b)
 
 	regInfo, err := h.recvRegistration(b)
 	if err != nil {
-		return nil, nil, false, common.NewBasicError("registration message error", nil, "err", err)
+		return nil, common.NewBasicError("registration message error", nil, "err", err)
 	}
-
-	tableEntry := newTableEntry()
-	ref, err := h.RoutingTable.Register(
-		regInfo.IA,
-		regInfo.PublicAddress,
-		getBindIP(regInfo.BindAddress),
-		regInfo.SVCAddress,
-		tableEntry,
-	)
+	appConn, _, err := appServer.Register(nil,
+		regInfo.IA, regInfo.PublicAddress, regInfo.SVCAddress)
 	if err != nil {
-		return nil, nil, false, common.NewBasicError("registration table error", nil, "err", err)
+		return nil, common.NewBasicError("registration table error", nil, "err", err)
 	}
-
-	udpRef := ref.(registration.RegReference)
-	port := uint16(udpRef.UDPAddr().Port)
+	udpAddr := appConn.(*dispatcher.Conn).LocalAddr().(*net.UDPAddr)
+	port := uint16(udpAddr.Port)
 	if err := h.sendConfirmation(b, &reliable.Confirmation{Port: port}); err != nil {
-		// Need to release stale state from the table
-		ref.Free()
-		return nil, nil, false, common.NewBasicError("confirmation message error", nil, "err", err)
+		appConn.Close()
+		return nil, common.NewBasicError("confirmation message error", nil, "err", err)
 	}
-	h.logRegistration(regInfo.IA, udpRef.UDPAddr(), getBindIP(regInfo.BindAddress),
+	h.logRegistration(regInfo.IA, udpAddr, getBindIP(regInfo.BindAddress),
 		regInfo.SVCAddress)
-	isIPv6 := regInfo.PublicAddress.IP.To4() == nil
-	return udpRef, tableEntry, isIPv6, nil
+	return appConn, nil
 }
 
 func (h *AppConnHandler) logRegistration(ia addr.IA, public *net.UDPAddr, bind net.IP,
@@ -197,8 +161,7 @@ func (h *AppConnHandler) sendConfirmation(b common.RawBytes, c *reliable.Confirm
 
 // RunAppToNetDataplane moves packets from the application's socket to the
 // overlay socket.
-func (h *AppConnHandler) RunAppToNetDataplane(ref registration.RegReference,
-	ovConn net.PacketConn) {
+func (h *AppConnHandler) RunAppToNetDataplane() {
 
 	for {
 		pkt := respool.GetPacket()
@@ -218,11 +181,7 @@ func (h *AppConnHandler) RunAppToNetDataplane(ref registration.RegReference,
 		metrics.M.AppReadBytes().Add(float64(pkt.Len()))
 		metrics.M.AppReadPkts().Inc()
 
-		if err := registerIfSCMPRequest(ref, &pkt.Info); err != nil {
-			log.Warn("SCMP Request ID error, packet still sent", "err", err)
-		}
-
-		n, err := pkt.SendOnConn(ovConn, pkt.OverlayRemote)
+		n, err := h.DispConn.Write(pkt)
 		if err != nil {
 			metrics.M.NetWriteErrors().Inc()
 			h.Logger.Error("[app->network] Overlay socket error", "err", err)
@@ -234,49 +193,31 @@ func (h *AppConnHandler) RunAppToNetDataplane(ref registration.RegReference,
 	}
 }
 
-// registerIfSCMPRequest registers the ID of the SCMP Request, if it is an
-// SCMP::General::EchoRequest, SCMP::General::TraceRouteRequest or SCMP::General::RecordPathRequest
-// packet. It also increments SCMP-related metrics.
-func registerIfSCMPRequest(ref registration.RegReference, packet *spkt.ScnPkt) error {
-	if scmpHdr, ok := packet.L4.(*scmp.Hdr); ok {
-		metrics.M.SCMPWritePkts(
-			metrics.SCMP{
-				Class: scmpHdr.Class.String(),
-				Type:  scmpHdr.Type.Name(scmpHdr.Class),
-			},
-		).Inc()
-		if !isSCMPGeneralRequest(scmpHdr) {
-			return nil
-		}
-		if id := getSCMPGeneralID(packet); id != 0 {
-			return ref.RegisterID(id)
-		}
-	}
-	return nil
-}
-
 // RunRingToAppDataplane moves packets from the application's ingress ring to
 // the application's socket.
-func (h *AppConnHandler) RunRingToAppDataplane(r *ringbuf.Ring) {
-	entries := make(ringbuf.EntryList, 1)
+func (h *AppConnHandler) RunRingToAppDataplane() {
 	for {
-		n, _ := r.Read(entries, true)
-		if n < 0 {
+		pkt := h.DispConn.Read()
+		if pkt == nil {
 			// Ring was closed because app shut down its data socket
 			return
 		}
-		if n > 0 {
-			pkt := entries[0].(*respool.Packet)
-			n, err := pkt.SendOnConn(h.Conn, pkt.OverlayRemote)
-			if err != nil {
-				metrics.M.AppWriteErrors().Inc()
-				h.Logger.Error("[network->app] App connection error.", "err", err)
-				h.Conn.Close()
-				return
-			}
-			metrics.M.AppWritePkts().Inc()
-			metrics.M.AppWriteBytes().Add(float64(n))
-			pkt.Free()
+		n, err := pkt.SendOnConn(h.Conn, pkt.OverlayRemote)
+		if err != nil {
+			metrics.M.AppWriteErrors().Inc()
+			h.Logger.Error("[network->app] App connection error.", "err", err)
+			h.Conn.Close()
+			return
 		}
+		metrics.M.AppWritePkts().Inc()
+		metrics.M.AppWriteBytes().Add(float64(n))
+		pkt.Free()
+	}
+}
+
+func getBindIP(address *net.UDPAddr) net.IP {
+	if address == nil {
+		return nil
 	}
+	return address.IP
 }
diff --git a/go/godispatcher/network/dispatcher.go b/go/godispatcher/network/dispatcher.go
index 984364f70..4a3aee334 100644
--- a/go/godispatcher/network/dispatcher.go
+++ b/go/godispatcher/network/dispatcher.go
@@ -15,54 +15,32 @@
 package network
 
 import (
-	"net"
 	"os"
-	"sync"
-	"time"
 
-	"github.com/scionproto/scion/go/godispatcher/internal/metrics"
+	"github.com/scionproto/scion/go/godispatcher/dispatcher"
 	"github.com/scionproto/scion/go/lib/common"
 	"github.com/scionproto/scion/go/lib/log"
-	"github.com/scionproto/scion/go/lib/overlay/conn"
 	"github.com/scionproto/scion/go/lib/sock/reliable"
 )
 
-// OverflowLoggingInterval is the minimum amount of time that needs to
-// pass before another overflow logging message is printed (if needed).
-const OverflowLoggingInterval = 10 * time.Second
-
-// ReceiveBufferSize is the size of receive buffers used by the dispatcher.
-const ReceiveBufferSize = 1 << 20
-
 type Dispatcher struct {
-	RoutingTable      *IATable
 	OverlaySocket     string
 	ApplicationSocket string
 	SocketFileMode    os.FileMode
 }
 
 func (d *Dispatcher) ListenAndServe() error {
-	metaLogger := &throttledMetaLogger{
-		Logger:      log.Root(),
-		MinInterval: OverflowLoggingInterval,
-	}
-	ipv4Conn, err := openConn("udp4", d.OverlaySocket, metaLogger)
-	if err != nil {
-		return err
-	}
-	defer ipv4Conn.Close()
-
-	ipv6Conn, err := openConn("udp6", d.OverlaySocket, metaLogger)
+	dispServer, err := dispatcher.NewServer(d.OverlaySocket)
 	if err != nil {
 		return err
 	}
-	defer ipv6Conn.Close()
+	defer dispServer.Close()
 
-	appServerConn, err := reliable.Listen(d.ApplicationSocket)
+	dispServerConn, err := reliable.Listen(d.ApplicationSocket)
 	if err != nil {
 		return err
 	}
-	defer appServerConn.Close()
+	defer dispServerConn.Close()
 	if err := os.Chmod(d.ApplicationSocket, d.SocketFileMode); err != nil {
 		return common.NewBasicError("chmod failed", err, "socket file", d.ApplicationSocket)
 	}
@@ -70,147 +48,17 @@ func (d *Dispatcher) ListenAndServe() error {
 	errChan := make(chan error)
 	go func() {
 		defer log.LogPanicAndExit()
-		netToRingDataplane := &NetToRingDataplane{
-			OverlayConn:  ipv4Conn,
-			RoutingTable: d.RoutingTable,
-		}
-		errChan <- netToRingDataplane.Run()
-	}()
-	go func() {
-		defer log.LogPanicAndExit()
-		netToRingDataplane := &NetToRingDataplane{
-			OverlayConn:  ipv6Conn,
-			RoutingTable: d.RoutingTable,
-		}
-		errChan <- netToRingDataplane.Run()
+		errChan <- dispServer.Serve()
 	}()
 
 	go func() {
 		defer log.LogPanicAndExit()
-		appServer := &AppSocketServer{
-			Listener: appServerConn,
-			ConnManager: &AppConnManager{
-				RoutingTable:    d.RoutingTable,
-				IPv4OverlayConn: ipv4Conn,
-				IPv6OverlayConn: ipv6Conn,
-			},
+		dispServer := &AppSocketServer{
+			Listener:   dispServerConn,
+			DispServer: dispServer,
 		}
-		errChan <- appServer.Serve()
+		errChan <- dispServer.Serve()
 	}()
 
 	return <-errChan
 }
-
-// openConn opens an overlay socket that tracks additional socket information
-// such as packets dropped due to buffer full.
-//
-// Note that Go-style dual-stacked IPv4/IPv6 connections are not supported. If
-// network is udp, it will be treated as udp4.
-func openConn(network, address string, p SocketMetaHandler) (net.PacketConn, error) {
-	// We cannot allow the Go standard library to open both types of sockets
-	// because the socket options are specific to only one socket type, so we
-	// degrade udp to only udp4.
-	if network == "udp" {
-		network = "udp4"
-	}
-	listeningAddress, err := net.ResolveUDPAddr(network, address)
-	if err != nil {
-		return nil, common.NewBasicError("unable to construct UDP addr", err)
-	}
-	if network == "udp4" && listeningAddress.IP == nil {
-		listeningAddress.IP = net.IPv4zero
-	}
-	if network == "udp6" && listeningAddress.IP == nil {
-		listeningAddress.IP = net.IPv6zero
-	}
-
-	c, err := conn.New(listeningAddress, nil, &conn.Config{ReceiveBufferSize: ReceiveBufferSize})
-	if err != nil {
-		return nil, common.NewBasicError("unable to open conn", err)
-	}
-
-	return &overlayConnWrapper{Conn: c, Handler: p}, nil
-}
-
-// SocketMetaHandler processes OS socket metadata during reads.
-type SocketMetaHandler interface {
-	Handle(*conn.ReadMeta)
-}
-
-// throttledMetaLogger logs packets dropped due to full receive buffers,
-// with a configurable threshold on how often logging messages are printed.
-type throttledMetaLogger struct {
-	// Logger is used to print the logging messages.
-	Logger log.Logger
-	// MinInterval is the minimum duration of time that has passed since the
-	MinInterval time.Duration
-
-	mu sync.Mutex
-	// lastPrintTimestamp is the time when the previous logging message was
-	// printed.
-	lastPrintTimestamp time.Time
-	// lastPrintValue is the overflow value printed in the last logging message.
-	lastPrintValue uint32
-}
-
-func (p *throttledMetaLogger) Handle(m *conn.ReadMeta) {
-	p.mu.Lock()
-	if m.RcvOvfl != p.lastPrintValue && time.Since(p.lastPrintTimestamp) > p.MinInterval {
-		if m.RcvOvfl > p.lastPrintValue {
-			metrics.M.NetReadOverflows().Add(float64(m.RcvOvfl - p.lastPrintValue))
-		} else {
-			metrics.M.NetReadOverflows().Add(float64(m.RcvOvfl))
-		}
-		p.Logger.Debug("Detected socket overflow", "total_cnt", m.RcvOvfl)
-		p.lastPrintTimestamp = time.Now()
-		p.lastPrintValue = m.RcvOvfl
-	}
-	p.mu.Unlock()
-}
-
-// overlayConnWrapper wraps a specialized overlay conn into a net.PacketConn
-// implementation. Only *net.UDPAddr addressing is supported.
-type overlayConnWrapper struct {
-	// Conn is the wrapped overlay connection object.
-	conn.Conn
-	// Handler is used to customize how the connection treats socket
-	// metadata.
-	Handler SocketMetaHandler
-}
-
-func (o *overlayConnWrapper) ReadFrom(p []byte) (int, net.Addr, error) {
-	n, meta, err := o.Conn.Read(common.RawBytes(p))
-	if meta == nil {
-		return n, nil, err
-	}
-	o.Handler.Handle(meta)
-	return n, meta.Src, err
-}
-
-func (o *overlayConnWrapper) WriteTo(p []byte, a net.Addr) (int, error) {
-	udpAddr, ok := a.(*net.UDPAddr)
-	if !ok {
-		return 0, common.NewBasicError("address is not UDP", nil, "addr", a)
-	}
-	return o.Conn.WriteTo(common.RawBytes(p), udpAddr)
-}
-
-func (o *overlayConnWrapper) Close() error {
-	return o.Conn.Close()
-}
-
-func (o *overlayConnWrapper) LocalAddr() net.Addr {
-	return o.Conn.LocalAddr()
-}
-
-func (o *overlayConnWrapper) SetDeadline(t time.Time) error {
-	return o.Conn.SetDeadline(t)
-}
-
-func (o *overlayConnWrapper) SetReadDeadline(t time.Time) error {
-	return o.Conn.SetReadDeadline(t)
-}
-
-func (o *overlayConnWrapper) SetWriteDeadline(t time.Time) error {
-	return o.Conn.SetWriteDeadline(t)
-}


Test patch:
diff --git a/go/godispatcher/network/overlay_test.go b/go/godispatcher/dispatcher/overlay_test.go
similarity index 99%
rename from go/godispatcher/network/overlay_test.go
rename to go/godispatcher/dispatcher/overlay_test.go
index 17f10bcf1..6e41ddddb 100644
--- a/go/godispatcher/network/overlay_test.go
+++ b/go/godispatcher/dispatcher/overlay_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package network
+package dispatcher
 
 import (
 	"net"