Merge pull request #610 from jbenet/reuseport-fix

p2p/net/conn: only reuseport if avail

Godeps/Godeps.json

@@ -160,7 +160,7 @@
 		},
 		{
 			"ImportPath": "github.com/jbenet/go-reuseport",
-			"Rev": "1e1968c4744fef51234e83f015aa0187b4bd796b"
+			"Rev": "a2e454f12a99b8898c41f9dcebae6c35dc3efa3a"
 		},
 		{
 			"ImportPath": "github.com/jbenet/go-sockaddr/net",

Godeps/_workspace/src/github.com/jbenet/go-reuseport/.travis.yml

 language: go
 
 go:
-  - 1.2
   - 1.3
   - 1.4
   - release
   - tip
 
 script:
-  - go test -v ./...
+  - go test -race -cpu=5 -v ./...

Godeps/_workspace/src/github.com/jbenet/go-reuseport/available_unix.go

+// +build darwin freebsd dragonfly netbsd openbsd linux
+package reuseport
+
+import (
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+)
+
+// checker is a struct to gather the availability check fields + funcs.
+// we use atomic ints because this is potentially a really hot function call.
+type checkerT struct {
+	avail int32      // atomic int managed by set/isAvailable()
+	check int32      // atomic int managed by has/checked()
+	mu    sync.Mutex // synchonizes the actual check
+}
+
+// the static location of the vars.
+var checker checkerT
+
+func (c *checkerT) isAvailable() bool {
+	return atomic.LoadInt32(&c.avail) != 0
+}
+
+func (c *checkerT) setIsAvailable(b bool) {
+	if b {
+		atomic.StoreInt32(&c.avail, 1)
+	} else {
+		atomic.StoreInt32(&c.avail, 0)
+	}
+}
+
+func (c *checkerT) hasChecked() bool {
+	return atomic.LoadInt32(&c.check) != 0
+}
+
+func (c *checkerT) setHasChecked(b bool) {
+	if b {
+		atomic.StoreInt32(&c.check, 1)
+	} else {
+		atomic.StoreInt32(&c.check, 0)
+	}
+}
+
+// Available returns whether or not SO_REUSEPORT is available in the OS.
+// It does so by attepting to open a tcp listener, setting the option, and
+// checking ENOPROTOOPT on error. After checking, the decision is cached
+// for the rest of the process run.
+func available() bool {
+	if checker.hasChecked() {
+		return checker.isAvailable()
+	}
+
+	// synchronize, only one should check
+	checker.mu.Lock()
+	defer checker.mu.Unlock()
+
+	// we blocked. someone may have been gotten this.
+	if checker.hasChecked() {
+		return checker.isAvailable()
+	}
+
+	// there may be fluke reasons to fail to add a listener.
+	// so we give it 5 shots. if not, give up and call it not avail.
+	for i := 0; i < 5; i++ {
+		// try to listen at tcp port 0.
+		l, err := listenStream("tcp", "127.0.0.1:0")
+		if err == nil {
+			// no error? available.
+			checker.setIsAvailable(true)
+			checker.setHasChecked(true)
+			l.Close() // Go back to the Shadow!
+			return true
+		}
+
+		if errno, ok := err.(syscall.Errno); ok {
+			if errno == syscall.ENOPROTOOPT {
+				break // :( that's all folks.
+			}
+		}
+
+		// not an errno? or not ENOPROTOOPT? retry.
+		<-time.After(20 * time.Millisecond) // wait a bit
+	}
+
+	checker.setIsAvailable(false)
+	checker.setHasChecked(true)
+	return false
+}

Godeps/_workspace/src/github.com/jbenet/go-reuseport/impl_windows.go

@@ -13,3 +13,10 @@ func listen(network, address string) (net.Listener, error) {
 func dial(dialer net.Dialer, network, address string) (net.Conn, error) {
 	return dialer.Dial(network, address)
 }
+
+// on windows, we just use the regular functions. sources
+// vary on this-- some claim port reuse behavior is on by default
+// on some windows systems. So we try. may the force be with you.
+func available() bool {
+	return true
+}

Godeps/_workspace/src/github.com/jbenet/go-reuseport/interface.go

@@ -20,9 +20,18 @@ package reuseport
 import (
 	"errors"
 	"net"
+	"syscall"
 	"time"
 )
 
+// Available returns whether or not SO_REUSEPORT is available in the OS.
+// It does so by attepting to open a tcp listener, setting the option, and
+// checking ENOPROTOOPT on error. After checking, the decision is cached
+// for the rest of the process run.
+func Available() bool {
+	return available()
+}
+
 // ErrUnsuportedProtocol signals that the protocol is not currently
 // supported by this package. This package currently only supports TCP.
 var ErrUnsupportedProtocol = errors.New("protocol not yet supported")
@@ -34,6 +43,10 @@ var ErrReuseFailed = errors.New("reuse failed")
 // Returns a net.Listener created from a file discriptor for a socket
 // with SO_REUSEPORT and SO_REUSEADDR option set.
 func Listen(network, address string) (net.Listener, error) {
+	if !available() {
+		return nil, syscall.Errno(syscall.ENOPROTOOPT)
+	}
+
 	return listenStream(network, address)
 }
 
@@ -41,6 +54,10 @@ func Listen(network, address string) (net.Listener, error) {
 // Returns a net.Listener created from a file discriptor for a socket
 // with SO_REUSEPORT and SO_REUSEADDR option set.
 func ListenPacket(network, address string) (net.PacketConn, error) {
+	if !available() {
+		return nil, syscall.Errno(syscall.ENOPROTOOPT)
+	}
+
 	return listenPacket(network, address)
 }
 
@@ -48,6 +65,9 @@ func ListenPacket(network, address string) (net.PacketConn, error) {
 // Returns a net.Conn created from a file discriptor for a socket
 // with SO_REUSEPORT and SO_REUSEADDR option set.
 func Dial(network, laddr, raddr string) (net.Conn, error) {
+	if !available() {
+		return nil, syscall.Errno(syscall.ENOPROTOOPT)
+	}
 
 	var d Dialer
 	if laddr != "" {

p2p/net/conn/dial.go

@@ -89,7 +89,7 @@ func (d *Dialer) rawConnDial(ctx context.Context, raddr ma.Multiaddr, remote pee
 	laddr := pickLocalAddr(d.LocalAddrs, raddr)
 	log.Debugf("%s dialing %s -- %s --> %s", d.LocalPeer, remote, laddr, raddr)
 
-	if laddr != nil {
+	if laddr != nil && reuseport.Available() {
 		// dial using reuseport.Dialer, because we're probably reusing addrs.
 		// this is optimistic, as the reuseDial may fail to bind the port.
 		if nconn, retry, reuseErr := d.reuseDial(laddr, raddr); reuseErr == nil {

p2p/net/conn/listen.go

@@ -124,20 +124,7 @@ func (l *listener) Loggable() map[string]interface{} {
 
 // Listen listens on the particular multiaddr, with given peer and peerstore.
 func Listen(ctx context.Context, addr ma.Multiaddr, local peer.ID, sk ic.PrivKey) (Listener, error) {
-
-	network, naddr, err := manet.DialArgs(addr)
-	if err != nil {
-		return nil, err
-	}
-
-	// _ := reuseport.Listen
-	// ml, err := manet.Listen(addr)
-	nl, err := reuseport.Listen(network, naddr)
-	if err != nil {
-		return nil, fmt.Errorf("Failed to listen on %s: %s", addr, err)
-	}
-
-	ml, err := manet.WrapNetListener(nl)
+	ml, err := manetListen(addr)
 	if err != nil {
 		return nil, err
 	}
@@ -154,3 +141,23 @@ func Listen(ctx context.Context, addr ma.Multiaddr, local peer.ID, sk ic.PrivKey
 	log.Event(ctx, "swarmListen", l)
 	return l, nil
 }
+
+func manetListen(addr ma.Multiaddr) (manet.Listener, error) {
+	network, naddr, err := manet.DialArgs(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	if reuseport.Available() {
+		nl, err := reuseport.Listen(network, naddr)
+		if err == nil {
+			// hey, it worked!
+			return manet.WrapNetListener(nl)
+		}
+		// reuseport is available, but we failed to listen. log debug, and retry normally.
+		log.Debugf("reuseport available, but failed to listen: %s %s, %s", network, naddr, err)
+	}
+
+	// either reuseport not available, or it failed. try normally.
+	return manet.Listen(addr)
+}