add in message type routing to the swarm object. tired, needs cleanup.
Jeromy committed
Aug 19, 2014 at 22:05 UTC
afdac2ca3abc6722b928d1833689752cfac4d30c
8 files changed
+167
-60
routing/dht/dht.go
+4
-3
@@ -100,10 +100,11 @@ func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
100
func (dht *IpfsDHT) handleMessages() {
101
u.DOut("Begin message handling routine\n")
102
103
- ch := dht.network.GetChan()
103
+ errs := dht.network.GetErrChan()
104
+ dhtmes := dht.network.GetChannel(swarm.PBWrapper_DHT_MESSAGE)
105
for {
106
select {
106
- case mes, ok := <-ch.Incoming:
107
+ case mes, ok := <-dhtmes:
108
if !ok {
109
u.DOut("handleMessages closing, bad recv on incoming\n")
110
return
@@ -147,7 +148,7 @@ func (dht *IpfsDHT) handleMessages() {
148
u.PErr("Recieved invalid message type")
149
}
150
150
- case err := <-ch.Errors:
151
+ case err := <-errs:
152
u.PErr("dht err: %s\n", err)
153
case <-dht.shutdown:
154
return
routing/dht/dht_test.go
+2
-2
@@ -132,8 +132,8 @@ func TestValueGetSet(t *testing.T) {
132
dhtA.Start()
133
dhtB.Start()
134
135
- errsa := dhtA.network.GetChan().Errors
136
- errsb := dhtB.network.GetChan().Errors
135
+ errsa := dhtA.network.GetErrChan()
136
+ errsb := dhtB.network.GetErrChan()
137
go func() {
138
select {
139
case err := <-errsa:
routing/dht/ext_test.go
+67
-3
@@ -66,8 +66,12 @@ func (f *fauxNet) Send(mes *swarm.Message) {
66
f.Chan.Outgoing <- mes
67
}
68
69
-func (f *fauxNet) GetChan() *swarm.Chan {
70
- return f.Chan
69
+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
75
}
76
77
func (f *fauxNet) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
@@ -167,7 +171,6 @@ func _randPeer() *peer.Peer {
171
}
172
173
func TestNotFound(t *testing.T) {
170
- u.Debug = true
174
fn := newFauxNet()
175
fn.Listen()
176
@@ -225,3 +228,64 @@ func TestNotFound(t *testing.T) {
228
}
229
t.Fatal("Expected to recieve an error.")
230
}
231
+
232
+// If less than K nodes are in the entire network, it should fail when we make
233
+// a GET rpc and nobody has the value
234
+func TestLessThanKResponses(t *testing.T) {
235
+ u.Debug = false
236
+ fn := newFauxNet()
237
+ fn.Listen()
238
+
239
+ local := new(peer.Peer)
240
+ local.ID = peer.ID("test_peer")
241
+
242
+ d := NewDHT(local, fn)
243
+ d.Start()
244
+
245
+ var ps []*peer.Peer
246
+ for i := 0; i < 5; i++ {
247
+ ps = append(ps, _randPeer())
248
+ d.Update(ps[i])
249
+ }
250
+ other := _randPeer()
251
+
252
+ // Reply with random peers to every message
253
+ fn.AddHandler(func(mes *swarm.Message) *swarm.Message {
254
+ t.Log("Handling message...")
255
+ pmes := new(PBDHTMessage)
256
+ err := proto.Unmarshal(mes.Data, pmes)
257
+ if err != nil {
258
+ t.Fatal(err)
259
+ }
260
+
261
+ switch pmes.GetType() {
262
+ case PBDHTMessage_GET_VALUE:
263
+ resp := Message{
264
+ Type: pmes.GetType(),
265
+ ID: pmes.GetId(),
266
+ Response: true,
267
+ Success: false,
268
+ Peers: []*peer.Peer{other},
269
+ }
270
+
271
+ return swarm.NewMessage(mes.Peer, resp.ToProtobuf())
272
+ default:
273
+ panic("Shouldnt recieve this.")
274
+ }
275
+
276
+ })
277
+
278
+ _, err := d.GetValue(u.Key("hello"), time.Second*30)
279
+ if err != nil {
280
+ switch err {
281
+ case u.ErrNotFound:
282
+ //Success!
283
+ return
284
+ case u.ErrTimeout:
285
+ t.Fatal("Should not have gotten timeout!")
286
+ default:
287
+ t.Fatalf("Got unexpected error: %s", err)
288
+ }
289
+ }
290
+ t.Fatal("Expected to recieve an error.")
291
+}
routing/dht/providers.go
+8
-8
@@ -3,24 +3,24 @@ package dht
3
import (
4
"time"
5
6
- u "github.com/jbenet/go-ipfs/util"
6
peer "github.com/jbenet/go-ipfs/peer"
7
+ u "github.com/jbenet/go-ipfs/util"
8
)
9
10
type ProviderManager struct {
11
providers map[u.Key][]*providerInfo
12
- newprovs chan *addProv
13
- getprovs chan *getProv
14
- halt chan struct{}
12
+ newprovs chan *addProv
13
+ getprovs chan *getProv
14
+ halt chan struct{}
15
}
16
17
type addProv struct {
18
- k u.Key
18
+ k u.Key
19
val *peer.Peer
20
}
21
22
type getProv struct {
23
- k u.Key
23
+ k u.Key
24
resp chan []*peer.Peer
25
}
26
@@ -55,7 +55,7 @@ func (pm *ProviderManager) run() {
55
for k, provs := range pm.providers {
56
var filtered []*providerInfo
57
for _, p := range provs {
58
- if time.Now().Sub(p.Creation) < time.Hour * 24 {
58
+ if time.Now().Sub(p.Creation) < time.Hour*24 {
59
filtered = append(filtered, p)
60
}
61
}
@@ -69,7 +69,7 @@ func (pm *ProviderManager) run() {
69
70
func (pm *ProviderManager) AddProvider(k u.Key, val *peer.Peer) {
71
pm.newprovs <- &addProv{
72
- k: k,
72
+ k: k,
73
val: val,
74
}
75
}
routing/dht/routing.go
+25
-26
@@ -164,7 +164,8 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
164
case p := <-npeerChan:
165
count++
166
if count >= KValue {
167
- break
167
+ errChan <- u.ErrNotFound
168
+ return
169
}
170
c.Increment()
171
@@ -172,40 +173,38 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
173
default:
174
if c.Size() == 0 {
175
errChan <- u.ErrNotFound
176
+ return
177
}
178
}
179
}
180
}()
181
182
process := func() {
181
- for {
182
- select {
183
- case p, ok := <-procPeer:
184
- if !ok || p == nil {
185
- c.Decrement()
186
- return
187
- }
188
- val, peers, err := dht.getValueOrPeers(p, key, timeout/4, routeLevel)
189
- if err != nil {
190
- u.DErr("%v\n", err.Error())
191
- c.Decrement()
192
- continue
193
- }
194
- if val != nil {
195
- valChan <- val
196
- c.Decrement()
197
- return
198
- }
183
+ for p := range procPeer {
184
+ if p == nil {
185
+ c.Decrement()
186
+ return
187
+ }
188
+ val, peers, err := dht.getValueOrPeers(p, key, timeout/4, routeLevel)
189
+ if err != nil {
190
+ u.DErr("%v\n", err.Error())
191
+ c.Decrement()
192
+ continue
193
+ }
194
+ if val != nil {
195
+ valChan <- val
196
+ c.Decrement()
197
+ return
198
+ }
199
200
- for _, np := range peers {
201
- // TODO: filter out peers that arent closer
202
- if !pset.Contains(np) && pset.Size() < KValue {
203
- pset.Add(np) //This is racey... make a single function to do operation
204
- npeerChan <- np
205
- }
200
+ for _, np := range peers {
201
+ // TODO: filter out peers that arent closer
202
+ if !pset.Contains(np) && pset.Size() < KValue {
203
+ pset.Add(np) //This is racey... make a single function to do operation
204
+ npeerChan <- np
205
}
207
- c.Decrement()
206
}
207
+ c.Decrement()
208
}
209
}
210
swarm/conn.go
-2
@@ -40,8 +40,6 @@ func Dial(network string, peer *peer.Peer) (*Conn, error) {
40
return nil, err
41
}
42
43
- fmt.Printf("Making connection to: %s\n", host)
44
-
43
nconn, err := net.Dial(network, host)
44
if err != nil {
45
return nil, err
swarm/interface.go
+2
-1
@@ -14,7 +14,8 @@ type Network interface {
14
Listen() error
15
ConnectNew(*ma.Multiaddr) (*peer.Peer, error)
16
GetConnection(id peer.ID, addr *ma.Multiaddr) (*peer.Peer, error)
17
- GetChan() *Chan
17
+ GetErrChan() chan error
18
+ GetChannel(PBWrapper_MessageType) chan *Message
19
Close()
20
Drop(*peer.Peer) error
21
}
swarm/swarm.go
+59
-15
@@ -84,6 +84,10 @@ type Swarm struct {
84
conns ConnMap
85
connsLock sync.RWMutex
86
87
+ filterChans map[PBWrapper_MessageType]chan *Message
88
+ toFilter chan *Message
89
+ newFilters chan *newFilterInfo
90
+
91
local *peer.Peer
92
listeners []net.Listener
93
}
@@ -91,10 +95,14 @@ type Swarm struct {
95
// NewSwarm constructs a Swarm, with a Chan.
96
func NewSwarm(local *peer.Peer) *Swarm {
97
s := &Swarm{
94
- Chan: NewChan(10),
95
- conns: ConnMap{},
96
- local: local,
97
- }
98
+ Chan: NewChan(10),
99
+ conns: ConnMap{},
100
+ local: local,
101
+ filterChans: make(map[PBWrapper_MessageType]chan *Message),
102
+ toFilter: make(chan *Message, 32),
103
+ newFilters: make(chan *newFilterInfo),
104
+ }
105
+ go s.routeMessages()
106
go s.fanOut()
107
return s
108
}
@@ -299,15 +307,8 @@ func (s *Swarm) fanIn(conn *Conn) {
307
goto out
308
}
309
302
- wrapper, err := Unwrap(data)
303
- if err != nil {
304
- s.Error(err)
305
- continue
306
- }
307
-
308
- // wrap it for consumers.
309
- msg := &Message{Peer: conn.Peer, Data: wrapper.GetMessage()}
310
- s.Chan.Incoming <- msg
310
+ msg := &Message{Peer: conn.Peer, Data: data}
311
+ s.toFilter <- msg
312
}
313
}
314
out:
@@ -317,6 +318,39 @@ out:
318
s.connsLock.Unlock()
319
}
320
321
+type newFilterInfo struct {
322
+ Type PBWrapper_MessageType
323
+ resp chan chan *Message
324
+}
325
+
326
+func (s *Swarm) routeMessages() {
327
+ for {
328
+ select {
329
+ case mes, ok := <-s.toFilter:
330
+ if !ok {
331
+ return
332
+ }
333
+ wrapper, err := Unwrap(mes.Data)
334
+ if err != nil {
335
+ u.PErr("error in route messages: %s\n", err)
336
+ }
337
+
338
+ ch, ok := s.filterChans[PBWrapper_MessageType(wrapper.GetType())]
339
+ if !ok {
340
+ u.PErr("Received message with invalid type: %d\n", wrapper.GetType())
341
+ continue
342
+ }
343
+
344
+ mes.Data = wrapper.GetMessage()
345
+ ch <- mes
346
+ case gchan := <-s.newFilters:
347
+ nch := make(chan *Message)
348
+ s.filterChans[gchan.Type] = nch
349
+ gchan.resp <- nch
350
+ }
351
+ }
352
+}
353
+
354
func (s *Swarm) Find(key u.Key) *peer.Peer {
355
s.connsLock.RLock()
356
defer s.connsLock.RUnlock()
@@ -414,8 +448,8 @@ func (s *Swarm) Error(e error) {
448
s.Chan.Errors <- e
449
}
450
417
-func (s *Swarm) GetChan() *Chan {
418
- return s.Chan
451
+func (s *Swarm) GetErrChan() chan error {
452
+ return s.Chan.Errors
453
}
454
455
func Wrap(data []byte, typ PBWrapper_MessageType) ([]byte, error) {
@@ -439,5 +473,15 @@ func Unwrap(data []byte) (*PBWrapper, error) {
473
return mes, nil
474
}
475
476
+func (s *Swarm) GetChannel(typ PBWrapper_MessageType) chan *Message {
477
+ nfi := &newFilterInfo{
478
+ Type: typ,
479
+ resp: make(chan chan *Message),
480
+ }
481
+ s.newFilters <- nfi
482
+
483
+ return <-nfi.resp
484
+}
485
+
486
// Temporary to ensure that the Swarm always matches the Network interface as we are changing it
487
var _ Network = &Swarm{}