p2p/net/swarm cleaned up dial sync
Juan Batiz-Benet committed
Jan 13, 2015 at 00:21 UTC
90d654d55c0729c53c1f99a5b0f90b4f7fe33349
2 files changed
+105
-32
p2p/net/swarm/swarm.go
+1
-7
@@ -4,7 +4,6 @@ package swarm
4
5
import (
6
"fmt"
7
- "sync"
7
8
inet "github.com/jbenet/go-ipfs/p2p/net"
9
addrutil "github.com/jbenet/go-ipfs/p2p/net/swarm/addr"
@@ -33,11 +32,7 @@ type Swarm struct {
32
local peer.ID
33
peers peer.Peerstore
34
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
35
+ dsync dialsync
36
37
cg ctxgroup.ContextGroup
38
}
@@ -59,7 +54,6 @@ func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr,
54
local: local,
55
peers: peers,
56
cg: ctxgroup.WithContext(ctx),
62
- dialing: map[peer.ID]chan struct{}{},
57
}
58
59
// configure Swarm
p2p/net/swarm/swarm_dial.go
+104
-25
@@ -3,6 +3,7 @@ package swarm
3
import (
4
"errors"
5
"fmt"
6
+ "sync"
7
8
conn "github.com/jbenet/go-ipfs/p2p/net/conn"
9
addrutil "github.com/jbenet/go-ipfs/p2p/net/swarm/addr"
@@ -13,6 +14,77 @@ import (
14
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
15
)
16
17
+// dialsync is a small object that helps manage ongoing dials.
18
+// this way, if we receive many simultaneous dial requests, one
19
+// can do its thing, while the rest wait.
20
+//
21
+// this interface is so would-be dialers can just:
22
+//
23
+// for {
24
+// c := findConnectionToPeer(peer)
25
+// if c != nil {
26
+// return c
27
+// }
28
+//
29
+// // ok, no connections. should we dial?
30
+// if ok, wait := dialsync.Lock(peer); !ok {
31
+// <-wait // can optionally wait
32
+// continue
33
+// }
34
+// defer dialsync.Unlock(peer)
35
+//
36
+// c := actuallyDial(peer)
37
+// return c
38
+// }
39
+//
40
+type dialsync struct {
41
+ // ongoing is a map of tickets for the current peers being dialed.
42
+ // this way, we dont kick off N dials simultaneously.
43
+ ongoing map[peer.ID]chan struct{}
44
+ lock sync.Mutex
45
+}
46
+
47
+// Lock governs the beginning of a dial attempt.
48
+// If there are no ongoing dials, it returns true, and the client is now
49
+// scheduled to dial. Every other goroutine that calls startDial -- with
50
+//the same dst -- will block until client is done. The client MUST call
51
+// ds.Unlock(p) when it is done, to unblock the other callers.
52
+// The client is not reponsible for achieving a successful dial, only for
53
+// reporting the end of the attempt (calling ds.Unlock(p)).
54
+//
55
+// see the example below `dialsync`
56
+func (ds *dialsync) Lock(dst peer.ID) (bool, chan struct{}) {
57
+ ds.lock.Lock()
58
+ if ds.ongoing == nil { // init if not ready
59
+ ds.ongoing = make(map[peer.ID]chan struct{})
60
+ }
61
+ wait, found := ds.ongoing[dst]
62
+ if !found {
63
+ ds.ongoing[dst] = make(chan struct{})
64
+ }
65
+ ds.lock.Unlock()
66
+
67
+ if found {
68
+ return false, wait
69
+ }
70
+
71
+ // ok! you're signed up to dial!
72
+ return true, nil
73
+}
74
+
75
+// Unlock releases waiters to a dial attempt. see Lock.
76
+// if Unlock(p) is called without calling Lock(p) first, Unlock panics.
77
+func (ds *dialsync) Unlock(dst peer.ID) {
78
+ ds.lock.Lock()
79
+ wait, found := ds.ongoing[dst]
80
+ if !found {
81
+ panic("called dialDone with no ongoing dials to peer: " + dst.Pretty())
82
+ }
83
+ delete(ds.ongoing, dst) // remove ongoing dial
84
+ close(wait) // release everyone else
85
+ ds.lock.Unlock()
86
+}
87
+
88
// Dial connects to a peer.
89
//
90
// The idea is that the client of Swarm does not need to know what network
@@ -20,46 +92,53 @@ import (
92
// This allows us to use various transport protocols, do NAT traversal/relay,
93
// etc. to achive connection.
94
func (s *Swarm) Dial(ctx context.Context, p peer.ID) (*Conn, error) {
23
-
95
if p == s.local {
96
return nil, errors.New("Attempted connection to self!")
97
}
98
28
- for {
99
+ // this loop is here because dials take time, and we should not be dialing
100
+ // the same peer concurrently (silly waste). Additonally, it's structured
101
+ // to check s.ConnectionsToPeer(p) _first_, and _between_ attempts because we
102
+ // may have received an incoming connection! if so, we no longer must dial.
103
+ //
104
+ // dial attempts. we may be doing the dialing. if not, we wait.
105
+ attempts := 3
106
+ var err error
107
+ var conn *Conn
108
+ for i := 0; i < attempts; i++ {
109
// check if we already have an open connection first
110
cs := s.ConnectionsToPeer(p)
31
- for _, c := range cs {
32
- if c != nil { // dump out the first one we find
33
- return c, nil
111
+ for _, conn = range cs {
112
+ if conn != nil { // dump out the first one we find. (TODO pick better)
113
+ return conn, nil
114
}
115
}
116
117
// 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 {
118
+ if ok, wait := s.dsync.Lock(p); !ok {
119
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():
120
+ case <-wait: // wait for that dial to finish.
121
+ continue // and see if it worked (loop), OR we got an incoming dial.
122
+ case <-ctx.Done(): // or we may have to bail...
123
return nil, ctx.Err()
124
}
125
}
126
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
127
+ // ok, we have been charged to dial! let's do it.
128
+ conn, err = s.dial(ctx, p)
129
+ s.dsync.Unlock(p)
130
+ if err != nil {
131
+ continue // ok, we failed. try again. (if loop is done, our error is output)
132
+ }
133
+ return conn, nil
134
+ }
135
+ return nil, err
136
+}
137
+
138
+// dial is the actual swarm's dial logic, gated by Dial.
139
+func (s *Swarm) dial(ctx context.Context, p peer.ID) (*Conn, error) {
140
+ if p == s.local {
141
+ return nil, errors.New("Attempted connection to self!")
142
}
143
144
sk := s.peers.PrivKey(s.local)