refactor dialing to not panic, and to be smart about ordering

whyrusleeping · whyrusleeping · commit 95381153b5f3 · 2015-10-13T00:47:35.000-07:00
License: MIT
Signed-off-by: Jeromy &lt;jeromyj@gmail.com&gt;
diff --git a/p2p/net/swarm/dial_test.go b/p2p/net/swarm/dial_test.go
@@ -2,6 +2,7 @@ package swarm
 
 import (
 	"net"
+	"sort"
 	"sync"
 	"testing"
 	"time"
@@ -438,3 +439,46 @@ func TestDialBackoffClears(t *testing.T) {
 		t.Log("correctly cleared backoff")
 	}
 }
+
+func mkAddr(t *testing.T, s string) ma.Multiaddr {
+	a, err := ma.NewMultiaddr(s)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return a
+}
+
+func TestDockerAddrCheck(t *testing.T) {
+	a := mkAddr(t, "/ip4/172.17.2.65/tcp/3124")
+
+	if !isDefaultDockerRange(a) {
+		t.Fatal("expected to be in docker range check")
+	}
+}
+
+func TestAddressSorting(t *testing.T) {
+	u1 := mkAddr(t, "/ip4/152.12.23.53/udp/1234/utp")
+	local := mkAddr(t, "/ip4/127.0.0.1/tcp/1234")
+	norm := mkAddr(t, "/ip4/6.5.4.3/tcp/1234")
+	docker := mkAddr(t, "/ip4/172.17.4.3/tcp/1234")
+
+	l := AddrList{local, docker, u1, norm}
+	sort.Sort(l)
+
+	if !l[0].Equal(u1) {
+		t.Fatal("expected utp addr to be sorted first")
+	}
+
+	if !l[1].Equal(local) {
+		t.Fatal("expected localhost addr second")
+	}
+
+	if !l[2].Equal(norm) {
+		t.Fatal("expected normal addr before docker addr")
+	}
+
+	if !l[3].Equal(docker) {
+		t.Fatal("expected docker addr last")
+	}
+}
diff --git a/p2p/net/swarm/swarm_dial.go b/p2p/net/swarm/swarm_dial.go
@@ -1,10 +1,12 @@
 package swarm
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
-	"math/rand"
 	"net"
+	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -16,9 +18,6 @@ import (
 
 	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 	manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
-	process "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
-	processctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
-	ratelimit "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/ratelimit"
 	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
 )
 
@@ -361,6 +360,9 @@ func (s *Swarm) dial(ctx context.Context, p peer.ID) (*Conn, error) {
 
 func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remoteAddrs []ma.Multiaddr) (conn.Conn, error) {
 
+	// sort addresses so preferred addresses are dialed sooner
+	sort.Sort(AddrList(remoteAddrs))
+
 	// try to connect to one of the peer's known addresses.
 	// we dial concurrently to each of the addresses, which:
 	// * makes the process faster overall
@@ -372,74 +374,77 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
 	defer cancel() // cancel work when we exit func
 
 	foundConn := make(chan struct{})
-	conns := make(chan conn.Conn, len(remoteAddrs))
+	conns := make(chan conn.Conn)
 	errs := make(chan error, len(remoteAddrs))
 
 	// dialSingleAddr is used in the rate-limited async thing below.
 	dialSingleAddr := func(addr ma.Multiaddr) {
+		// rebind chans in scope so we can nil them out easily
+		connsout := conns
+		errsout := errs
+
 		connC, err := s.dialAddr(ctx, d, p, addr)
+		if err != nil {
+			connsout = nil
+		} else if connC == nil {
+			// NOTE: this really should never happen
+			log.Errorf("failed to dial %s %s and got no error!", p, addr)
+			err = fmt.Errorf("failed to dial %s %s", p, addr)
+			connsout = nil
+		} else {
+			errsout = nil
+		}
 
 		// check parent still wants our results
 		select {
 		case <-foundConn:
 			if connC != nil {
 				connC.Close()
 			}
-			return
-		default:
-		}
-
-		if err != nil {
-			errs <- err
-		} else if connC == nil {
-			errs <- fmt.Errorf("failed to dial %s %s", p, addr)
-		} else {
-			conns <- connC
+		case errsout <- err:
+		case connsout <- connC:
 		}
 	}
 
 	// this whole thing is in a goroutine so we can use foundConn
 	// to end early.
 	go func() {
-		// rate limiting just in case. at most 10 addrs at once.
-		limiter := ratelimit.NewRateLimiter(process.Background(), 8)
-		limiter.Go(func(worker process.Process) {
-			// permute addrs so we try different sets first each time.
-			for _, i := range rand.Perm(len(remoteAddrs)) {
-				select {
-				case <-foundConn: // if one of them succeeded already
-					break
-				case <-worker.Closing(): // our context was cancelled
-					break
-				default:
-				}
-
-				workerAddr := remoteAddrs[i] // shadow variable to avoid race
-
-				// we have to do the waiting concurrently because there are addrs
-				// that SHOULD NOT be rate limited (utp), nor blocked by other
-				// rate limited addrs (tcp).
-				//
-				// (and we need to call `limiter.Go`, instead of `go` as required
-				// by goproc/limiter semantics. note: limiter.Go is not LimitedGo.)
-				limiter.Go(func(p process.Process) {
-
-					// returns whatever ratelimiting is acceptable for workerAddr.
-					// may not rate limit at all.
-					rl := s.addrDialRateLimit(workerAddr)
-					rl <- struct{}{}
-
-					limiter.LimitedGo(func(worker process.Process) {
-						dialSingleAddr(workerAddr)
-					})
-
-					<-rl
-				})
+		limiter := make(chan struct{}, 8)
+		// permute addrs so we try different sets first each time.
+		for _, addr := range remoteAddrs {
+			select {
+			case <-foundConn: // if one of them succeeded already
+				return
+			case <-ctx.Done(): // our context was cancelled
+				return
+			case limiter <- struct{}{}:
+				// continue
+			}
 
+			// returns whatever ratelimiting is acceptable for workerAddr.
+			// may not rate limit at all.
+			rl := s.addrDialRateLimit(addr)
+			select {
+			case <-foundConn: // if one of them succeeded already
+				return
+			case <-ctx.Done(): // our context was cancelled
+				return
+			case rl <- struct{}{}:
+				// continue
 			}
-		})
 
-		processctx.CloseAfterContext(limiter, ctx)
+			// we have to do the waiting concurrently because there are addrs
+			// that SHOULD NOT be rate limited (utp), nor blocked by other
+			// rate limited addrs (tcp).
+			go func(rlc <-chan struct{}, a ma.Multiaddr) {
+				defer func() {
+					<-limiter
+					<-rlc
+				}()
+				dialSingleAddr(a)
+			}(rl, addr)
+
+		}
 	}()
 
 	// wair fot the results.
@@ -533,3 +538,63 @@ func isTCPMultiaddr(a ma.Multiaddr) bool {
 	p := a.Protocols()
 	return len(p) == 2 && (p[0].Name == "ip4" || p[0].Name == "ip6") && p[1].Name == "tcp"
 }
+
+func isDefaultDockerRange(a ma.Multiaddr) bool {
+	parts := strings.Split(a.String(), "/")
+	if len(parts) != 5 {
+		return false
+	}
+
+	if parts[1] == "ip4" && strings.HasPrefix(parts[2], "172.17.") {
+		return true
+	}
+
+	return false
+}
+
+type AddrList []ma.Multiaddr
+
+func (al AddrList) Len() int {
+	return len(al)
+}
+
+func (al AddrList) Swap(i, j int) {
+	al[i], al[j] = al[j], al[i]
+}
+
+func (al AddrList) Less(i, j int) bool {
+	a := al[i]
+	b := al[j]
+
+	// dial utp and similar 'non-fd-consuming' addresses first
+	if !isFDCostlyTransport(a) {
+		if isFDCostlyTransport(b) {
+			return true
+		}
+
+		// if neither consume fd's, assume equal ordering
+		return false
+	}
+
+	// dial localhost addresses next, they should fail immediately
+	if manet.IsIPLoopback(a) {
+		if !manet.IsIPLoopback(b) {
+			return true
+		}
+
+		// both local? equal
+		return false
+	}
+
+	// docker addresses should be tried last. they very rarely work.
+	if isDefaultDockerRange(a) {
+		return false
+	}
+
+	if isDefaultDockerRange(b) {
+		return true
+	}
+
+	// for the rest, just sort by bytes
+	return bytes.Compare(a.Bytes(), b.Bytes()) > 0
+}
