fix swarm message type code, i beleive it works well now
Jeromy committed
Aug 20, 2014 at 16:51 UTC
05b80afc3576e72f2fa471833f5851f615ea441a
9 files changed
+93
-69
routing/dht/dht.go
+13
-12
@@ -28,6 +28,7 @@ type IpfsDHT struct {
28
routingTables []*kb.RoutingTable
29
30
network swarm.Network
31
+ netChan *swarm.Chan
32
33
// Local peer (yourself)
34
self *peer.Peer
@@ -55,6 +56,7 @@ type IpfsDHT struct {
56
func NewDHT(p *peer.Peer, net swarm.Network) *IpfsDHT {
57
dht := new(IpfsDHT)
58
dht.network = net
59
+ dht.netChan = net.GetChannel(swarm.PBWrapper_DHT_MESSAGE)
60
dht.datastore = ds.NewMapDatastore()
61
dht.self = p
62
dht.providers = NewProviderManager()
@@ -101,10 +103,9 @@ func (dht *IpfsDHT) handleMessages() {
103
u.DOut("Begin message handling routine\n")
104
105
errs := dht.network.GetErrChan()
104
- dhtmes := dht.network.GetChannel(swarm.PBWrapper_DHT_MESSAGE)
106
for {
107
select {
107
- case mes, ok := <-dhtmes:
108
+ case mes, ok := <-dht.netChan.Incoming:
109
if !ok {
110
u.DOut("handleMessages closing, bad recv on incoming\n")
111
return
@@ -165,7 +166,7 @@ func (dht *IpfsDHT) putValueToNetwork(p *peer.Peer, key string, value []byte) er
166
}
167
168
mes := swarm.NewMessage(p, pmes.ToProtobuf())
168
- dht.network.Send(mes)
169
+ dht.netChan.Outgoing <- mes
170
return nil
171
}
172
@@ -225,7 +226,7 @@ func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *PBDHTMessage) {
226
227
out:
228
mes := swarm.NewMessage(p, resp.ToProtobuf())
228
- dht.network.Send(mes)
229
+ dht.netChan.Outgoing <- mes
230
}
231
232
// Store a value in this peer local storage
@@ -247,7 +248,7 @@ func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *PBDHTMessage) {
248
ID: pmes.GetId(),
249
}
250
250
- dht.network.Send(swarm.NewMessage(p, resp.ToProtobuf()))
251
+ dht.netChan.Outgoing <- swarm.NewMessage(p, resp.ToProtobuf())
252
}
253
254
func (dht *IpfsDHT) handleFindPeer(p *peer.Peer, pmes *PBDHTMessage) {
@@ -258,7 +259,7 @@ func (dht *IpfsDHT) handleFindPeer(p *peer.Peer, pmes *PBDHTMessage) {
259
}
260
defer func() {
261
mes := swarm.NewMessage(p, resp.ToProtobuf())
261
- dht.network.Send(mes)
262
+ dht.netChan.Outgoing <- mes
263
}()
264
level := pmes.GetValue()[0]
265
u.DOut("handleFindPeer: searching for '%s'\n", peer.ID(pmes.GetKey()).Pretty())
@@ -310,7 +311,7 @@ func (dht *IpfsDHT) handleGetProviders(p *peer.Peer, pmes *PBDHTMessage) {
311
}
312
313
mes := swarm.NewMessage(p, resp.ToProtobuf())
313
- dht.network.Send(mes)
314
+ dht.netChan.Outgoing <- mes
315
}
316
317
type providerInfo struct {
@@ -336,7 +337,7 @@ func (dht *IpfsDHT) handleDiagnostic(p *peer.Peer, pmes *PBDHTMessage) {
337
338
for _, ps := range seq {
339
mes := swarm.NewMessage(ps, pmes)
339
- dht.network.Send(mes)
340
+ dht.netChan.Outgoing <- mes
341
}
342
343
buf := new(bytes.Buffer)
@@ -372,7 +373,7 @@ out:
373
}
374
375
mes := swarm.NewMessage(p, resp.ToProtobuf())
375
- dht.network.Send(mes)
376
+ dht.netChan.Outgoing <- mes
377
}
378
379
func (dht *IpfsDHT) getValueOrPeers(p *peer.Peer, key u.Key, timeout time.Duration, level int) ([]byte, []*peer.Peer, error) {
@@ -429,7 +430,7 @@ func (dht *IpfsDHT) getValueSingle(p *peer.Peer, key u.Key, timeout time.Duratio
430
431
mes := swarm.NewMessage(p, pmes.ToProtobuf())
432
t := time.Now()
432
- dht.network.Send(mes)
433
+ dht.netChan.Outgoing <- mes
434
435
// Wait for either the response or a timeout
436
timeup := time.After(timeout)
@@ -545,7 +546,7 @@ func (dht *IpfsDHT) findPeerSingle(p *peer.Peer, id peer.ID, timeout time.Durati
546
mes := swarm.NewMessage(p, pmes.ToProtobuf())
547
listenChan := dht.listener.Listen(pmes.ID, 1, time.Minute)
548
t := time.Now()
548
- dht.network.Send(mes)
549
+ dht.netChan.Outgoing <- mes
550
after := time.After(timeout)
551
select {
552
case <-after:
@@ -581,7 +582,7 @@ func (dht *IpfsDHT) findProvidersSingle(p *peer.Peer, key u.Key, level int, time
582
mes := swarm.NewMessage(p, pmes.ToProtobuf())
583
584
listenChan := dht.listener.Listen(pmes.ID, 1, time.Minute)
584
- dht.network.Send(mes)
585
+ dht.netChan.Outgoing <- mes
586
after := time.After(timeout)
587
select {
588
case <-after:
routing/dht/ext_test.go
+2
-2
@@ -70,8 +70,8 @@ func (f *fauxNet) GetErrChan() chan error {
70
return f.Chan.Errors
71
}
72
73
-func (f *fauxNet) GetChannel(t swarm.PBWrapper_MessageType) chan *swarm.Message {
74
- return f.Chan.Incoming
73
+func (f *fauxNet) GetChannel(t swarm.PBWrapper_MessageType) *swarm.Chan {
74
+ return f.Chan
75
}
76
77
func (f *fauxNet) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
routing/dht/routing.go
+3
-3
@@ -240,7 +240,7 @@ func (dht *IpfsDHT) Provide(key u.Key) error {
240
241
for _, p := range peers {
242
mes := swarm.NewMessage(p, pbmes)
243
- dht.network.Send(mes)
243
+ dht.netChan.Outgoing <- mes
244
}
245
return nil
246
}
@@ -352,7 +352,7 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
352
353
before := time.Now()
354
responseChan := dht.listener.Listen(pmes.ID, 1, time.Minute)
355
- dht.network.Send(mes)
355
+ dht.netChan.Outgoing <- mes
356
357
tout := time.After(timeout)
358
select {
@@ -385,7 +385,7 @@ func (dht *IpfsDHT) getDiagnostic(timeout time.Duration) ([]*diagInfo, error) {
385
pbmes := pmes.ToProtobuf()
386
for _, p := range targets {
387
mes := swarm.NewMessage(p, pbmes)
388
- dht.network.Send(mes)
388
+ dht.netChan.Outgoing <- mes
389
}
390
391
var out []*diagInfo
swarm/interface.go
+2
-3
@@ -8,14 +8,13 @@ import (
8
)
9
10
type Network interface {
11
- Send(*Message)
12
- Error(error)
11
Find(u.Key) *peer.Peer
12
Listen() error
13
ConnectNew(*ma.Multiaddr) (*peer.Peer, error)
14
GetConnection(id peer.ID, addr *ma.Multiaddr) (*peer.Peer, error)
15
+ Error(error)
16
GetErrChan() chan error
18
- GetChannel(PBWrapper_MessageType) chan *Message
17
+ GetChannel(PBWrapper_MessageType) *Chan
18
Close()
19
Drop(*peer.Peer) error
20
}
swarm/mes_wrapper.pb.go
+7
-4
@@ -23,14 +23,17 @@ var _ = math.Inf
23
type PBWrapper_MessageType int32
24
25
const (
26
- PBWrapper_DHT_MESSAGE PBWrapper_MessageType = 0
26
+ PBWrapper_TEST PBWrapper_MessageType = 0
27
+ PBWrapper_DHT_MESSAGE PBWrapper_MessageType = 1
28
)
29
30
var PBWrapper_MessageType_name = map[int32]string{
30
- 0: "DHT_MESSAGE",
31
+ 0: "TEST",
32
+ 1: "DHT_MESSAGE",
33
}
34
var PBWrapper_MessageType_value = map[string]int32{
33
- "DHT_MESSAGE": 0,
35
+ "TEST": 0,
36
+ "DHT_MESSAGE": 1,
37
}
38
39
func (x PBWrapper_MessageType) Enum() *PBWrapper_MessageType {
@@ -64,7 +67,7 @@ func (m *PBWrapper) GetType() PBWrapper_MessageType {
67
if m != nil && m.Type != nil {
68
return *m.Type
69
}
67
- return PBWrapper_DHT_MESSAGE
70
+ return PBWrapper_TEST
71
}
72
73
func (m *PBWrapper) GetMessage() []byte {
swarm/mes_wrapper.proto
+2
-1
@@ -2,7 +2,8 @@ package swarm;
2
3
message PBWrapper {
4
enum MessageType {
5
- DHT_MESSAGE = 0;
5
+ TEST = 0;
6
+ DHT_MESSAGE = 1;
7
}
8
9
required MessageType Type = 1;
swarm/swarm.go
+34
-41
@@ -38,7 +38,7 @@ func NewMessage(p *peer.Peer, data proto.Message) *Message {
38
}
39
}
40
41
-// Chan is a swam channel, which provides duplex communication and errors.
41
+// Chan is a swarm channel, which provides duplex communication and errors.
42
type Chan struct {
43
Outgoing chan *Message
44
Incoming chan *Message
@@ -84,7 +84,7 @@ type Swarm struct {
84
conns ConnMap
85
connsLock sync.RWMutex
86
87
- filterChans map[PBWrapper_MessageType]chan *Message
87
+ filterChans map[PBWrapper_MessageType]*Chan
88
toFilter chan *Message
89
newFilters chan *newFilterInfo
90
@@ -98,7 +98,7 @@ func NewSwarm(local *peer.Peer) *Swarm {
98
Chan: NewChan(10),
99
conns: ConnMap{},
100
local: local,
101
- filterChans: make(map[PBWrapper_MessageType]chan *Message),
101
+ filterChans: make(map[PBWrapper_MessageType]*Chan),
102
toFilter: make(chan *Message, 32),
103
newFilters: make(chan *newFilterInfo),
104
}
@@ -233,6 +233,8 @@ func (s *Swarm) Dial(peer *peer.Peer) (*Conn, error, bool) {
233
return conn, nil, false
234
}
235
236
+// StartConn adds the passed in connection to its peerMap and starts
237
+// the fanIn routine for that connection
238
func (s *Swarm) StartConn(conn *Conn) error {
239
if conn == nil {
240
return errors.New("Tried to start nil connection.")
@@ -275,14 +277,8 @@ func (s *Swarm) fanOut() {
277
continue
278
}
279
278
- wrapped, err := Wrap(msg.Data, PBWrapper_DHT_MESSAGE)
279
- if err != nil {
280
- s.Error(err)
281
- continue
282
- }
283
-
280
// queue it in the connection's buffer
285
- conn.Outgoing.MsgChan <- wrapped
281
+ conn.Outgoing.MsgChan <- msg.Data
282
}
283
}
284
}
@@ -320,7 +316,7 @@ out:
316
317
type newFilterInfo struct {
318
Type PBWrapper_MessageType
323
- resp chan chan *Message
319
+ resp chan *Chan
320
}
321
322
func (s *Swarm) routeMessages() {
@@ -342,15 +338,36 @@ func (s *Swarm) routeMessages() {
338
}
339
340
mes.Data = wrapper.GetMessage()
345
- ch <- mes
341
+ ch.Incoming <- mes
342
case gchan := <-s.newFilters:
347
- nch := make(chan *Message)
348
- s.filterChans[gchan.Type] = nch
343
+ nch, ok := s.filterChans[gchan.Type]
344
+ if !ok {
345
+ nch = NewChan(16)
346
+ s.filterChans[gchan.Type] = nch
347
+ go s.muxChan(nch, gchan.Type)
348
+ }
349
gchan.resp <- nch
350
}
351
}
352
}
353
354
+func (s *Swarm) muxChan(ch *Chan, typ PBWrapper_MessageType) {
355
+ for {
356
+ select {
357
+ case <-ch.Close:
358
+ return
359
+ case mes := <-ch.Outgoing:
360
+ data, err := Wrap(mes.Data, typ)
361
+ if err != nil {
362
+ u.PErr("muxChan error: %s\n", err)
363
+ continue
364
+ }
365
+ mes.Data = data
366
+ s.Chan.Outgoing <- mes
367
+ }
368
+ }
369
+}
370
+
371
func (s *Swarm) Find(key u.Key) *peer.Peer {
372
s.connsLock.RLock()
373
defer s.connsLock.RUnlock()
@@ -386,6 +403,7 @@ func (s *Swarm) GetConnection(id peer.ID, addr *ma.Multiaddr) (*peer.Peer, error
403
return conn.Peer, err
404
}
405
406
+// Handle performing a handshake on a new connection and ensuring proper forward communication
407
func (s *Swarm) handleDialedCon(conn *Conn) error {
408
err := ident.Handshake(s.local, conn.Peer, conn.Incoming.MsgChan, conn.Outgoing.MsgChan)
409
if err != nil {
@@ -440,10 +458,6 @@ func (s *Swarm) Drop(p *peer.Peer) error {
458
return conn.Close()
459
}
460
443
-func (s *Swarm) Send(mes *Message) {
444
- s.Chan.Outgoing <- mes
445
-}
446
-
461
func (s *Swarm) Error(e error) {
462
s.Chan.Errors <- e
463
}
@@ -452,31 +466,10 @@ func (s *Swarm) GetErrChan() chan error {
466
return s.Chan.Errors
467
}
468
455
-func Wrap(data []byte, typ PBWrapper_MessageType) ([]byte, error) {
456
- wrapper := new(PBWrapper)
457
- wrapper.Message = data
458
- wrapper.Type = &typ
459
- b, err := proto.Marshal(wrapper)
460
- if err != nil {
461
- return nil, err
462
- }
463
- return b, nil
464
-}
465
-
466
-func Unwrap(data []byte) (*PBWrapper, error) {
467
- mes := new(PBWrapper)
468
- err := proto.Unmarshal(data, mes)
469
- if err != nil {
470
- return nil, err
471
- }
472
-
473
- return mes, nil
474
-}
475
-
476
-func (s *Swarm) GetChannel(typ PBWrapper_MessageType) chan *Message {
469
+func (s *Swarm) GetChannel(typ PBWrapper_MessageType) *Chan {
470
nfi := &newFilterInfo{
471
Type: typ,
479
- resp: make(chan chan *Message),
472
+ resp: make(chan *Chan),
473
}
474
s.newFilters <- nfi
475
swarm/swarm_test.go
+6
-3
@@ -38,7 +38,8 @@ func pong(c net.Conn, peer *peer.Peer) {
38
fmt.Printf("error: didn't receive ping: '%v'\n", b.GetMessage())
39
return
40
}
41
- data, err = Wrap([]byte("pong"), PBWrapper_DHT_MESSAGE)
41
+
42
+ data, err = Wrap([]byte("pong"), PBWrapper_TEST)
43
if err != nil {
44
fmt.Printf("error %v\n", err)
45
return
@@ -63,6 +64,7 @@ func TestSwarm(t *testing.T) {
64
"11140beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33": "/ip4/127.0.0.1/tcp/4567",
65
}
66
67
+ recv := swarm.GetChannel(PBWrapper_TEST)
68
for k, n := range peerNames {
69
peer, err := setupPeer(k, n)
70
if err != nil {
@@ -96,13 +98,14 @@ func TestSwarm(t *testing.T) {
98
MsgNum := 1000
99
for k := 0; k < MsgNum; k++ {
100
for _, p := range peers {
99
- swarm.Chan.Outgoing <- &Message{Peer: p, Data: []byte("ping")}
101
+ recv.Outgoing <- &Message{Peer: p, Data: []byte("ping")}
102
}
103
}
104
105
got := map[u.Key]int{}
106
+
107
for k := 0; k < (MsgNum * len(peers)); k++ {
105
- msg := <-swarm.Chan.Incoming
108
+ msg := <-recv.Incoming
109
if string(msg.Data) != "pong" {
110
t.Error("unexpected conn output", msg.Data)
111
}
swarm/wrapper.go
new
+24
@@ -0,0 +1,24 @@
1
+package swarm
2
+
3
+import "code.google.com/p/goprotobuf/proto"
4
+
5
+func Wrap(data []byte, typ PBWrapper_MessageType) ([]byte, error) {
6
+ wrapper := new(PBWrapper)
7
+ wrapper.Message = data
8
+ wrapper.Type = &typ
9
+ b, err := proto.Marshal(wrapper)
10
+ if err != nil {
11
+ return nil, err
12
+ }
13
+ return b, nil
14
+}
15
+
16
+func Unwrap(data []byte) (*PBWrapper, error) {
17
+ mes := new(PBWrapper)
18
+ err := proto.Unmarshal(data, mes)
19
+ if err != nil {
20
+ return nil, err
21
+ }
22
+
23
+ return mes, nil
24
+}