all: make conn.Bind.Open return a slice of receive functions

Instead of hard-coding exactly two sources from which
to receive packets (an IPv4 source and an IPv6 source),
allow the conn.Bind to specify a set of sources.

Beneficial consequences:

* If there's no IPv6 support on a system,
  conn.Bind.Open can choose not to return a receive function for it,
  which is simpler than tracking that state in the bind.
  This simplification removes existing data races from both
  conn.StdNetBind and bindtest.ChannelBind.
* If there are more than two sources on a system,
  the conn.Bind no longer needs to add a separate muxing layer.

Signed-off-by: Josh Bleecher Snyder <[email protected]>

Author: josharian

conn/bind_linux.go

@@ -55,10 +55,11 @@ func (endpoint *LinuxSocketEndpoint) dst6() *unix.SockaddrInet6 {
 
 // LinuxSocketBind uses sendmsg and recvmsg to implement a full bind with sticky sockets on Linux.
 type LinuxSocketBind struct {
-	sock4    int
-	sock6    int
-	lastMark uint32
-	closing  sync.RWMutex
+	// mu guards sock4 and sock6 and the associated fds.
+	// As long as someone holds mu (read or write), the associated fds are valid.
+	mu    sync.RWMutex
+	sock4 int
+	sock6 int
 }
 
 func NewLinuxSocketBind() Bind { return &LinuxSocketBind{sock4: -1, sock6: -1} }
@@ -102,54 +103,67 @@ func (*LinuxSocketBind) ParseEndpoint(s string) (Endpoint, error) {
 	return nil, errors.New("invalid IP address")
 }
 
-func (bind *LinuxSocketBind) Open(port uint16) (uint16, error) {
+func (bind *LinuxSocketBind) Open(port uint16) ([]ReceiveFunc, uint16, error) {
+	bind.mu.Lock()
+	defer bind.mu.Unlock()
+
 	var err error
 	var newPort uint16
 	var tries int
 
 	if bind.sock4 != -1 || bind.sock6 != -1 {
-		return 0, ErrBindAlreadyOpen
+		return nil, 0, ErrBindAlreadyOpen
 	}
 
 	originalPort := port
 
 again:
 	port = originalPort
+	var sock4, sock6 int
 	// Attempt ipv6 bind, update port if successful.
-	bind.sock6, newPort, err = create6(port)
+	sock6, newPort, err = create6(port)
 	if err != nil {
-		if err != syscall.EAFNOSUPPORT {
-			return 0, err
+		if !errors.Is(err, syscall.EAFNOSUPPORT) {
+			return nil, 0, err
 		}
 	} else {
 		port = newPort
 	}
 
 	// Attempt ipv4 bind, update port if successful.
-	bind.sock4, newPort, err = create4(port)
+	sock4, newPort, err = create4(port)
 	if err != nil {
-		if originalPort == 0 && err == syscall.EADDRINUSE && tries < 100 {
-			unix.Close(bind.sock6)
+		if originalPort == 0 && errors.Is(err, syscall.EADDRINUSE) && tries < 100 {
+			unix.Close(sock6)
 			tries++
 			goto again
 		}
-		if err != syscall.EAFNOSUPPORT {
-			unix.Close(bind.sock6)
-			return 0, err
+		if !errors.Is(err, syscall.EAFNOSUPPORT) {
+			unix.Close(sock6)
+			return nil, 0, err
 		}
 	} else {
 		port = newPort
 	}
 
-	if bind.sock4 == -1 && bind.sock6 == -1 {
-		return 0, syscall.EAFNOSUPPORT
+	var fns []ReceiveFunc
+	if sock4 != -1 {
+		fns = append(fns, makeReceiveIPv4(sock4))
+		bind.sock4 = sock4
+	}
+	if sock6 != -1 {
+		fns = append(fns, makeReceiveIPv6(sock6))
+		bind.sock6 = sock6
+	}
+	if len(fns) == 0 {
+		return nil, 0, syscall.EAFNOSUPPORT
 	}
-	return port, nil
+	return fns, port, nil
 }
 
 func (bind *LinuxSocketBind) SetMark(value uint32) error {
-	bind.closing.RLock()
-	defer bind.closing.RUnlock()
+	bind.mu.RLock()
+	defer bind.mu.RUnlock()
 
 	if bind.sock6 != -1 {
 		err := unix.SetsockoptInt(
@@ -177,21 +191,24 @@ func (bind *LinuxSocketBind) SetMark(value uint32) error {
 		}
 	}
 
-	bind.lastMark = value
 	return nil
 }
 
 func (bind *LinuxSocketBind) Close() error {
-	var err1, err2 error
-	bind.closing.RLock()
+	// Take a readlock to shut down the sockets...
+	bind.mu.RLock()
 	if bind.sock6 != -1 {
 		unix.Shutdown(bind.sock6, unix.SHUT_RDWR)
 	}
 	if bind.sock4 != -1 {
 		unix.Shutdown(bind.sock4, unix.SHUT_RDWR)
 	}
-	bind.closing.RUnlock()
-	bind.closing.Lock()
+	bind.mu.RUnlock()
+	// ...and a write lock to close the fd.
+	// This ensures that no one else is using the fd.
+	bind.mu.Lock()
+	defer bind.mu.Unlock()
+	var err1, err2 error
 	if bind.sock6 != -1 {
 		err1 = unix.Close(bind.sock6)
 		bind.sock6 = -1
@@ -200,54 +217,36 @@ func (bind *LinuxSocketBind) Close() error {
 		err2 = unix.Close(bind.sock4)
 		bind.sock4 = -1
 	}
-	bind.closing.Unlock()
 
 	if err1 != nil {
 		return err1
 	}
 	return err2
 }
 
-func (bind *LinuxSocketBind) ReceiveIPv6(buff []byte) (int, Endpoint, error) {
-	bind.closing.RLock()
-	defer bind.closing.RUnlock()
-
-	var end LinuxSocketEndpoint
-	if bind.sock6 == -1 {
-		return 0, nil, net.ErrClosed
+func makeReceiveIPv6(sock int) ReceiveFunc {
+	return func(buff []byte) (int, Endpoint, error) {
+		var end LinuxSocketEndpoint
+		n, err := receive6(sock, buff, &end)
+		return n, &end, err
 	}
-	n, err := receive6(
-		bind.sock6,
-		buff,
-		&end,
-	)
-	return n, &end, err
 }
 
-func (bind *LinuxSocketBind) ReceiveIPv4(buff []byte) (int, Endpoint, error) {
-	bind.closing.RLock()
-	defer bind.closing.RUnlock()
-
-	var end LinuxSocketEndpoint
-	if bind.sock4 == -1 {
-		return 0, nil, net.ErrClosed
+func makeReceiveIPv4(sock int) ReceiveFunc {
+	return func(buff []byte) (int, Endpoint, error) {
+		var end LinuxSocketEndpoint
+		n, err := receive4(sock, buff, &end)
+		return n, &end, err
 	}
-	n, err := receive4(
-		bind.sock4,
-		buff,
-		&end,
-	)
-	return n, &end, err
 }
 
 func (bind *LinuxSocketBind) Send(buff []byte, end Endpoint) error {
-	bind.closing.RLock()
-	defer bind.closing.RUnlock()
-
 	nend, ok := end.(*LinuxSocketEndpoint)
 	if !ok {
 		return ErrWrongEndpointType
 	}
+	bind.mu.RLock()
+	defer bind.mu.RUnlock()
 	if !nend.isV6 {
 		if bind.sock4 == -1 {
 			return net.ErrClosed

conn/bind_std.go

@@ -8,6 +8,7 @@ package conn
 import (
 	"errors"
 	"net"
+	"sync"
 	"syscall"
 )
 
@@ -16,6 +17,7 @@ import (
 // It uses the Go's net package to implement networking.
 // See LinuxSocketBind for a proper implementation on the Linux platform.
 type StdNetBind struct {
+	mu         sync.Mutex // protects following fields
 	ipv4       *net.UDPConn
 	ipv6       *net.UDPConn
 	blackhole4 bool
@@ -81,12 +83,15 @@ func listenNet(network string, port int) (*net.UDPConn, int, error) {
 	return conn, uaddr.Port, nil
 }
 
-func (bind *StdNetBind) Open(uport uint16) (uint16, error) {
+func (bind *StdNetBind) Open(uport uint16) ([]ReceiveFunc, uint16, error) {
+	bind.mu.Lock()
+	defer bind.mu.Unlock()
+
 	var err error
 	var tries int
 
 	if bind.ipv4 != nil || bind.ipv6 != nil {
-		return 0, ErrBindAlreadyOpen
+		return nil, 0, ErrBindAlreadyOpen
 	}
 
 	// Attempt to open ipv4 and ipv6 listeners on the same port.
@@ -97,7 +102,7 @@ again:
 
 	ipv4, port, err = listenNet("udp4", port)
 	if err != nil && !errors.Is(err, syscall.EAFNOSUPPORT) {
-		return 0, err
+		return nil, 0, err
 	}
 
 	// Listen on the same port as we're using for ipv4.
@@ -109,17 +114,27 @@ again:
 	}
 	if err != nil && !errors.Is(err, syscall.EAFNOSUPPORT) {
 		ipv4.Close()
-		return 0, err
+		return nil, 0, err
 	}
-	if ipv4 == nil && ipv6 == nil {
-		return 0, syscall.EAFNOSUPPORT
+	var fns []ReceiveFunc
+	if ipv4 != nil {
+		fns = append(fns, makeReceiveFunc(ipv4, true))
+		bind.ipv4 = ipv4
 	}
-	bind.ipv4 = ipv4
-	bind.ipv6 = ipv6
-	return uint16(port), nil
+	if ipv6 != nil {
+		fns = append(fns, makeReceiveFunc(ipv6, false))
+		bind.ipv6 = ipv6
+	}
+	if len(fns) == 0 {
+		return nil, 0, syscall.EAFNOSUPPORT
+	}
+	return fns, uint16(port), nil
 }
 
 func (bind *StdNetBind) Close() error {
+	bind.mu.Lock()
+	defer bind.mu.Unlock()
+
 	var err1, err2 error
 	if bind.ipv4 != nil {
 		err1 = bind.ipv4.Close()
@@ -137,23 +152,14 @@ func (bind *StdNetBind) Close() error {
 	return err2
 }
 
-func (bind *StdNetBind) ReceiveIPv4(buff []byte) (int, Endpoint, error) {
-	if bind.ipv4 == nil {
-		return 0, nil, syscall.EAFNOSUPPORT
+func makeReceiveFunc(conn *net.UDPConn, isIPv4 bool) ReceiveFunc {
+	return func(buff []byte) (int, Endpoint, error) {
+		n, endpoint, err := conn.ReadFromUDP(buff)
+		if isIPv4 && endpoint != nil {
+			endpoint.IP = endpoint.IP.To4()
+		}
+		return n, (*StdNetEndpoint)(endpoint), err
 	}
-	n, endpoint, err := bind.ipv4.ReadFromUDP(buff)
-	if endpoint != nil {
-		endpoint.IP = endpoint.IP.To4()
-	}
-	return n, (*StdNetEndpoint)(endpoint), err
-}
-
-func (bind *StdNetBind) ReceiveIPv6(buff []byte) (int, Endpoint, error) {
-	if bind.ipv6 == nil {
-		return 0, nil, syscall.EAFNOSUPPORT
-	}
-	n, endpoint, err := bind.ipv6.ReadFromUDP(buff)
-	return n, (*StdNetEndpoint)(endpoint), err
 }
 
 func (bind *StdNetBind) Send(buff []byte, endpoint Endpoint) error {
@@ -162,15 +168,16 @@ func (bind *StdNetBind) Send(buff []byte, endpoint Endpoint) error {
 	if !ok {
 		return ErrWrongEndpointType
 	}
-	var conn *net.UDPConn
-	var blackhole bool
-	if nend.IP.To4() != nil {
-		blackhole = bind.blackhole4
-		conn = bind.ipv4
-	} else {
+
+	bind.mu.Lock()
+	blackhole := bind.blackhole4
+	conn := bind.ipv4
+	if nend.IP.To4() == nil {
 		blackhole = bind.blackhole6
 		conn = bind.ipv6
 	}
+	bind.mu.Unlock()
+
 	if blackhole {
 		return nil
 	}