p2p/net/swarm: dial once at a time
Juan Batiz-Benet committed
Jan 12, 2015 at 22:14 UTC
fcece3e32e54daea7ae940ad7eece0df23b60564
3 files changed
+88
-9
p2p/net/swarm/simul_test.go
+43
@@ -11,6 +11,49 @@ import (
11
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
12
)
13
14
+func TestSimultDials(t *testing.T) {
15
+ // t.Skip("skipping for another test")
16
+
17
+ ctx := context.Background()
18
+ swarms := makeSwarms(ctx, t, 2)
19
+
20
+ // connect everyone
21
+ {
22
+ var wg sync.WaitGroup
23
+ connect := func(s *Swarm, dst peer.ID, addr ma.Multiaddr) {
24
+ // copy for other peer
25
+ log.Debugf("TestSimultOpen: connecting: %s --> %s (%s)", s.local, dst, addr)
26
+ s.peers.AddAddress(dst, addr)
27
+ if _, err := s.Dial(ctx, dst); err != nil {
28
+ t.Fatal("error swarm dialing to peer", err)
29
+ }
30
+ wg.Done()
31
+ }
32
+
33
+ log.Info("Connecting swarms simultaneously.")
34
+ for i := 0; i < 10; i++ { // connect 10x for each.
35
+ wg.Add(2)
36
+ go connect(swarms[0], swarms[1].local, swarms[1].ListenAddresses()[0])
37
+ go connect(swarms[1], swarms[0].local, swarms[0].ListenAddresses()[0])
38
+ }
39
+ wg.Wait()
40
+ }
41
+
42
+ // should still just have 1, at most 2 connections :)
43
+ c01l := len(swarms[0].ConnectionsToPeer(swarms[1].local))
44
+ if c01l > 2 {
45
+ t.Error("0->1 has", c01l)
46
+ }
47
+ c10l := len(swarms[1].ConnectionsToPeer(swarms[0].local))
48
+ if c10l > 2 {
49
+ t.Error("1->0 has", c10l)
50
+ }
51
+
52
+ for _, s := range swarms {
53
+ s.Close()
54
+ }
55
+}
56
+
57
func TestSimultOpen(t *testing.T) {
58
// t.Skip("skipping for another test")
59
p2p/net/swarm/swarm.go
+11
-4
@@ -4,6 +4,7 @@ package swarm
4
5
import (
6
"fmt"
7
+ "sync"
8
9
inet "github.com/jbenet/go-ipfs/p2p/net"
10
addrutil "github.com/jbenet/go-ipfs/p2p/net/swarm/addr"
@@ -33,6 +34,11 @@ type Swarm struct {
34
peers peer.Peerstore
35
connh ConnHandler
36
37
+ // dialing is a channel for the current peers being dialed.
38
+ // this way, we dont kick off N dials simultaneously.
39
+ dialing map[peer.ID]chan struct{}
40
+ dialingmu sync.Mutex
41
+
42
cg ctxgroup.ContextGroup
43
}
44
@@ -49,10 +55,11 @@ func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr,
55
}
56
57
s := &Swarm{
52
- swarm: ps.NewSwarm(PSTransport),
53
- local: local,
54
- peers: peers,
55
- cg: ctxgroup.WithContext(ctx),
58
+ swarm: ps.NewSwarm(PSTransport),
59
+ local: local,
60
+ peers: peers,
61
+ cg: ctxgroup.WithContext(ctx),
62
+ dialing: map[peer.ID]chan struct{}{},
63
}
64
65
// configure Swarm
p2p/net/swarm/swarm_dial.go
+34
-5
@@ -25,12 +25,41 @@ func (s *Swarm) Dial(ctx context.Context, p peer.ID) (*Conn, error) {
25
return nil, errors.New("Attempted connection to self!")
26
}
27
28
- // check if we already have an open connection first
29
- cs := s.ConnectionsToPeer(p)
30
- for _, c := range cs {
31
- if c != nil { // dump out the first one we find
32
- return c, nil
28
+ for {
29
+ // check if we already have an open connection first
30
+ cs := s.ConnectionsToPeer(p)
31
+ for _, c := range cs {
32
+ if c != nil { // dump out the first one we find
33
+ return c, nil
34
+ }
35
}
36
+
37
+ // check if there's an ongoing dial to this peer
38
+ s.dialingmu.Lock()
39
+ dialDone, found := s.dialing[p]
40
+ if !found { // if not, set one up.
41
+ dialDone = make(chan struct{})
42
+ s.dialing[p] = dialDone
43
+ }
44
+ s.dialingmu.Unlock()
45
+
46
+ if found {
47
+ select {
48
+ case <-dialDone: // wait for that dial to finish.
49
+ continue // and see if it worked (loop). it may not have.
50
+ case <-ctx.Done():
51
+ return nil, ctx.Err()
52
+ }
53
+ }
54
+
55
+ // else, we're the ones dialing for others.
56
+ defer func() {
57
+ s.dialingmu.Lock()
58
+ delete(s.dialing, p)
59
+ close(dialDone)
60
+ s.dialingmu.Unlock()
61
+ }()
62
+ break
63
}
64
65
sk := s.peers.PrivKey(s.local)