modify use of swarm to not make duplicate connections
Jeromy committed
Aug 12, 2014 at 15:37 UTC
b8a6fbbf7d46c8ef9b26dafea712ac0cc0462541
8 files changed
+76
-29
identify/identify.go
+1
-1
@@ -14,7 +14,7 @@ func Handshake(self, remote *peer.Peer, in, out chan []byte) error {
14
out <- self.ID
15
resp := <-in
16
remote.ID = peer.ID(resp)
17
- u.DOut("identify: Got node id: %s", remote.ID.Pretty())
17
+ u.DOut("[%s] identify: Got node id: %s", self.ID.Pretty(), remote.ID.Pretty())
18
19
return nil
20
}
routing/dht/DHTMessage.go
+2
@@ -28,6 +28,8 @@ func peerInfo(p *peer.Peer) *PBDHTMessage_PBPeer {
28
return pbp
29
}
30
31
+// TODO: building the protobuf message this way is a little wasteful
32
+// Unused fields wont be omitted, find a better way to do this
33
func (m *DHTMessage) ToProtobuf() *PBDHTMessage {
34
pmes := new(PBDHTMessage)
35
if m.Value != nil {
routing/dht/dht.go
+2
-2
@@ -93,7 +93,7 @@ func (dht *IpfsDHT) Start() {
93
func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
94
maddrstr, _ := addr.String()
95
u.DOut("Connect to new peer: %s", maddrstr)
96
- npeer, err := dht.network.Connect(addr)
96
+ npeer, err := dht.network.ConnectNew(addr)
97
if err != nil {
98
return nil, err
99
}
@@ -527,7 +527,7 @@ func (dht *IpfsDHT) getFromPeerList(key u.Key, timeout time.Duration,
527
continue
528
}
529
530
- p, err = dht.network.Connect(maddr)
530
+ p, err = dht.network.GetConnection(peer.ID(pinfo.GetId()), maddr)
531
if err != nil {
532
u.PErr("getValue error: %s", err)
533
continue
routing/dht/routing.go
+20
-6
@@ -3,6 +3,7 @@ package dht
3
import (
4
"bytes"
5
"encoding/json"
6
+ "errors"
7
"math/rand"
8
"time"
9
@@ -89,10 +90,10 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
90
panic("not yet implemented")
91
}
92
92
- p, err = s.network.Connect(maddr)
93
+ p, err = s.network.GetConnection(peer.ID(closers[0].GetId()), maddr)
94
if err != nil {
94
- // Move up route level
95
- panic("not yet implemented.")
95
+ u.PErr("[%s] Failed to connect to: %s", s.self.ID.Pretty(), closers[0].GetAddr())
96
+ route_level++
97
}
98
} else {
99
route_level++
@@ -160,12 +161,13 @@ func (s *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer,
161
for _, prov := range pmes_out.GetPeers() {
162
p := s.network.Find(u.Key(prov.GetId()))
163
if p == nil {
164
+ u.DOut("given provider %s was not in our network already.", peer.ID(prov.GetId()).Pretty())
165
maddr, err := ma.NewMultiaddr(prov.GetAddr())
166
if err != nil {
167
u.PErr("error connecting to new peer: %s", err)
168
continue
169
}
168
- p, err = s.network.Connect(maddr)
170
+ p, err = s.network.GetConnection(peer.ID(prov.GetId()), maddr)
171
if err != nil {
172
u.PErr("error connecting to new peer: %s", err)
173
continue
@@ -183,11 +185,20 @@ func (s *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer,
185
186
// FindPeer searches for a peer with given ID.
187
func (s *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, error) {
188
+ // Check if were already connected to them
189
+ p, _ := s.Find(id)
190
+ if p != nil {
191
+ return p, nil
192
+ }
193
+
194
route_level := 0
187
- p := s.routes[route_level].NearestPeer(kb.ConvertPeerID(id))
195
+ p = s.routes[route_level].NearestPeer(kb.ConvertPeerID(id))
196
if p == nil {
197
return nil, kb.ErrLookupFailure
198
}
199
+ if p.ID.Equal(id) {
200
+ return p, nil
201
+ }
202
203
for route_level < len(s.routes) {
204
pmes, err := s.findPeerSingle(p, id, timeout, route_level)
@@ -202,11 +213,14 @@ func (s *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, error
213
return nil, u.WrapError(err, "FindPeer received bad info")
214
}
215
205
- nxtPeer, err := s.network.Connect(addr)
216
+ nxtPeer, err := s.network.GetConnection(peer.ID(found.GetId()), addr)
217
if err != nil {
218
return nil, u.WrapError(err, "FindPeer failed to connect to new peer.")
219
}
220
if pmes.GetSuccess() {
221
+ if !id.Equal(nxtPeer.ID) {
222
+ return nil, errors.New("got back invalid peer from 'successful' response")
223
+ }
224
return nxtPeer, nil
225
} else {
226
p = nxtPeer
swarm/conn.go
+2
-1
@@ -2,11 +2,12 @@ package swarm
2
3
import (
4
"fmt"
5
+ "net"
6
+
7
peer "github.com/jbenet/go-ipfs/peer"
8
u "github.com/jbenet/go-ipfs/util"
9
msgio "github.com/jbenet/go-msgio"
10
ma "github.com/jbenet/go-multiaddr"
9
- "net"
11
)
12
13
// ChanBuffer is the size of the buffer in the Conn Chan
swarm/interface.go
+2
-1
@@ -12,7 +12,8 @@ type Network interface {
12
Error(error)
13
Find(u.Key) *peer.Peer
14
Listen() error
15
- Connect(*ma.Multiaddr) (*peer.Peer, error)
15
+ ConnectNew(*ma.Multiaddr) (*peer.Peer, error)
16
+ GetConnection(id peer.ID, addr *ma.Multiaddr) (*peer.Peer, error)
17
GetChan() *Chan
18
Close()
19
Drop(*peer.Peer) error
swarm/swarm.go
+42
-15
@@ -203,7 +203,7 @@ func (s *Swarm) Close() {
203
// etc. to achive connection.
204
//
205
// For now, Dial uses only TCP. This will be extended.
206
-func (s *Swarm) Dial(peer *peer.Peer) (*Conn, error) {
206
+func (s *Swarm) Dial(peer *peer.Peer) (*Conn, error, bool) {
207
k := peer.Key()
208
209
// check if we already have an open connection first
@@ -211,17 +211,16 @@ func (s *Swarm) Dial(peer *peer.Peer) (*Conn, error) {
211
conn, found := s.conns[k]
212
s.connsLock.RUnlock()
213
if found {
214
- return conn, nil
214
+ return conn, nil, true
215
}
216
217
// open connection to peer
218
conn, err := Dial("tcp", peer)
219
if err != nil {
220
- return nil, err
220
+ return nil, err, false
221
}
222
223
- s.StartConn(conn)
224
- return conn, nil
223
+ return conn, nil, false
224
}
225
226
func (s *Swarm) StartConn(conn *Conn) error {
@@ -309,35 +308,63 @@ func (s *Swarm) Find(key u.Key) *peer.Peer {
308
return conn.Peer
309
}
310
312
-func (s *Swarm) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
313
- if addr == nil {
314
- return nil, errors.New("nil Multiaddr passed to swarm.Connect()")
311
+// GetConnection will check if we are already connected to the peer in question
312
+// and only open a new connection if we arent already
313
+func (s *Swarm) GetConnection(id peer.ID, addr *ma.Multiaddr) (*peer.Peer, error) {
314
+ p := &peer.Peer{
315
+ ID: id,
316
+ Addresses: []*ma.Multiaddr{addr},
317
}
316
- npeer := new(peer.Peer)
317
- npeer.AddAddress(addr)
318
319
- conn, err := Dial("tcp", npeer)
319
+ conn, err, reused := s.Dial(p)
320
if err != nil {
321
return nil, err
322
}
323
324
- err = ident.Handshake(s.local, npeer, conn.Incoming.MsgChan, conn.Outgoing.MsgChan)
324
+ if reused {
325
+ return p, nil
326
+ }
327
+
328
+ err = s.handleDialedCon(conn)
329
+ return conn.Peer, err
330
+}
331
+
332
+func (s *Swarm) handleDialedCon(conn *Conn) error {
333
+ err := ident.Handshake(s.local, conn.Peer, conn.Incoming.MsgChan, conn.Outgoing.MsgChan)
334
if err != nil {
326
- return nil, err
335
+ return err
336
}
337
338
// Send node an address that you can be reached on
339
myaddr := s.local.NetAddress("tcp")
340
mastr, err := myaddr.String()
341
if err != nil {
333
- return nil, errors.New("No local address to send to peer.")
342
+ errors.New("No local address to send to peer.")
343
}
344
345
conn.Outgoing.MsgChan <- []byte(mastr)
346
347
s.StartConn(conn)
348
340
- return npeer, nil
349
+ return nil
350
+}
351
+
352
+// ConnectNew is for connecting to a peer when you dont know their ID,
353
+// Should only be used when you are sure that you arent already connected to peer in question
354
+func (s *Swarm) ConnectNew(addr *ma.Multiaddr) (*peer.Peer, error) {
355
+ if addr == nil {
356
+ return nil, errors.New("nil Multiaddr passed to swarm.Connect()")
357
+ }
358
+ npeer := new(peer.Peer)
359
+ npeer.AddAddress(addr)
360
+
361
+ conn, err := Dial("tcp", npeer)
362
+ if err != nil {
363
+ return nil, err
364
+ }
365
+
366
+ err = s.handleDialedCon(conn)
367
+ return npeer, err
368
}
369
370
// Removes a given peer from the swarm and closes connections to it
swarm/swarm_test.go
+5
-3
@@ -2,11 +2,12 @@ package swarm
2
3
import (
4
"fmt"
5
+ "net"
6
+ "testing"
7
+
8
peer "github.com/jbenet/go-ipfs/peer"
9
u "github.com/jbenet/go-ipfs/util"
10
msgio "github.com/jbenet/go-msgio"
8
- "net"
9
- "testing"
11
)
12
13
func pingListen(listener *net.TCPListener, peer *peer.Peer) {
@@ -71,11 +72,12 @@ func TestSwarm(t *testing.T) {
72
}
73
go pingListen(listener.(*net.TCPListener), peer)
74
74
- _, err = swarm.Dial(peer)
75
+ conn, err, _ := swarm.Dial(peer)
76
if err != nil {
77
t.Fatal("error swarm dialing to peer", err)
78
}
79
80
+ swarm.StartConn(conn)
81
// ok done, add it.
82
peers = append(peers, peer)
83
listeners = append(listeners, listener)