fixed small bug introduced during race condition frustration
Jeromy committed
Aug 7, 2014 at 14:16 UTC
01ca93b4f5adfefe1348a51d1860011a4c0c02f4
3 files changed
+148
-32
routing/dht/dht.go
+85
-15
@@ -3,6 +3,7 @@ package dht
3
import (
4
"sync"
5
"time"
6
+ "bytes"
7
"encoding/json"
8
9
peer "github.com/jbenet/go-ipfs/peer"
@@ -22,7 +23,7 @@ import (
23
// IpfsDHT is an implementation of Kademlia with Coral and S/Kademlia modifications.
24
// It is used to implement the base IpfsRouting module.
25
type IpfsDHT struct {
25
- routes *RoutingTable
26
+ routes []*RoutingTable
27
28
network *swarm.Swarm
29
@@ -38,7 +39,7 @@ type IpfsDHT struct {
39
providerLock sync.RWMutex
40
41
// map of channels waiting for reply messages
41
- listeners map[uint64]chan *swarm.Message
42
+ listeners map[uint64]*listenInfo
43
listenLock sync.RWMutex
44
45
// Signal to shutdown dht
@@ -46,6 +47,14 @@ type IpfsDHT struct {
47
48
// When this peer started up
49
birth time.Time
50
+
51
+ //lock to make diagnostics work better
52
+ diaglock sync.Mutex
53
+}
54
+
55
+type listenInfo struct {
56
+ resp chan *swarm.Message
57
+ count int
58
}
59
60
// Create a new DHT object with the given peer as the 'local' host
@@ -63,10 +72,11 @@ func NewDHT(p *peer.Peer) (*IpfsDHT, error) {
72
dht.network = network
73
dht.datastore = ds.NewMapDatastore()
74
dht.self = p
66
- dht.listeners = make(map[uint64]chan *swarm.Message)
75
+ dht.listeners = make(map[uint64]*listenInfo)
76
dht.providers = make(map[u.Key][]*providerInfo)
77
dht.shutdown = make(chan struct{})
69
- dht.routes = NewRoutingTable(20, convertPeerID(p.ID))
78
+ dht.routes = make([]*RoutingTable, 1)
79
+ dht.routes[0] = NewRoutingTable(20, convertPeerID(p.ID))
80
dht.birth = time.Now()
81
return dht, nil
82
}
@@ -106,7 +116,7 @@ func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) (*peer.Peer, error) {
116
117
dht.network.StartConn(conn)
118
109
- removed := dht.routes.Update(peer)
119
+ removed := dht.routes[0].Update(peer)
120
if removed != nil {
121
panic("need to remove this peer.")
122
}
@@ -142,7 +152,7 @@ func (dht *IpfsDHT) handleMessages() {
152
}
153
154
// Update peers latest visit in routing table
145
- removed := dht.routes.Update(mes.Peer)
155
+ removed := dht.routes[0].Update(mes.Peer)
156
if removed != nil {
157
panic("Need to handle removed peer.")
158
}
@@ -150,10 +160,15 @@ func (dht *IpfsDHT) handleMessages() {
160
// Note: not sure if this is the correct place for this
161
if pmes.GetResponse() {
162
dht.listenLock.RLock()
153
- ch, ok := dht.listeners[pmes.GetId()]
163
+ list, ok := dht.listeners[pmes.GetId()]
164
+ if list.count > 1 {
165
+ list.count--
166
+ } else if list.count == 1 {
167
+ delete(dht.listeners, pmes.GetId())
168
+ }
169
dht.listenLock.RUnlock()
170
if ok {
156
- ch <- mes
171
+ list.resp <- mes
172
} else {
173
// this is expected behaviour during a timeout
174
u.DOut("Received response with nobody listening...")
@@ -181,7 +196,7 @@ func (dht *IpfsDHT) handleMessages() {
196
case DHTMessage_PING:
197
dht.handlePing(mes.Peer, pmes)
198
case DHTMessage_DIAGNOSTIC:
184
- // TODO: network diagnostic messages
199
+ dht.handleDiagnostic(mes.Peer, pmes)
200
}
201
202
case err := <-dht.network.Chan.Errors:
@@ -220,7 +235,7 @@ func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *DHTMessage) {
235
}
236
} else if err == ds.ErrNotFound {
237
// Find closest peer(s) to desired key and reply with that info
223
- closer := dht.routes.NearestPeer(convertKey(u.Key(pmes.GetKey())))
238
+ closer := dht.routes[0].NearestPeer(convertKey(u.Key(pmes.GetKey())))
239
resp = &pDHTMessage{
240
Response: true,
241
Id: *pmes.Id,
@@ -256,7 +271,7 @@ func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *DHTMessage) {
271
272
func (dht *IpfsDHT) handleFindPeer(p *peer.Peer, pmes *DHTMessage) {
273
u.POut("handleFindPeer: searching for '%s'", peer.ID(pmes.GetKey()).Pretty())
259
- closest := dht.routes.NearestPeer(convertKey(u.Key(pmes.GetKey())))
274
+ closest := dht.routes[0].NearestPeer(convertKey(u.Key(pmes.GetKey())))
275
if closest == nil {
276
panic("could not find anything.")
277
}
@@ -336,10 +351,10 @@ func (dht *IpfsDHT) handleAddProvider(p *peer.Peer, pmes *DHTMessage) {
351
352
// Register a handler for a specific message ID, used for getting replies
353
// to certain messages (i.e. response to a GET_VALUE message)
339
-func (dht *IpfsDHT) ListenFor(mesid uint64) <-chan *swarm.Message {
354
+func (dht *IpfsDHT) ListenFor(mesid uint64, count int) <-chan *swarm.Message {
355
lchan := make(chan *swarm.Message)
356
dht.listenLock.Lock()
342
- dht.listeners[mesid] = lchan
357
+ dht.listeners[mesid] = &listenInfo{lchan, count}
358
dht.listenLock.Unlock()
359
return lchan
360
}
@@ -347,12 +362,19 @@ func (dht *IpfsDHT) ListenFor(mesid uint64) <-chan *swarm.Message {
362
// Unregister the given message id from the listener map
363
func (dht *IpfsDHT) Unlisten(mesid uint64) {
364
dht.listenLock.Lock()
350
- ch, ok := dht.listeners[mesid]
365
+ list, ok := dht.listeners[mesid]
366
if ok {
367
delete(dht.listeners, mesid)
368
}
369
dht.listenLock.Unlock()
355
- close(ch)
370
+ close(list.resp)
371
+}
372
+
373
+func (dht *IpfsDHT) IsListening(mesid uint64) bool {
374
+ dht.listenLock.RLock()
375
+ _,ok := dht.listeners[mesid]
376
+ dht.listenLock.RUnlock()
377
+ return ok
378
}
379
380
// Stop all communications from this peer and shut down
@@ -368,3 +390,51 @@ func (dht *IpfsDHT) addProviderEntry(key u.Key, p *peer.Peer) {
390
dht.providers[key] = append(provs, &providerInfo{time.Now(), p})
391
dht.providerLock.Unlock()
392
}
393
+
394
+func (dht *IpfsDHT) handleDiagnostic(p *peer.Peer, pmes *DHTMessage) {
395
+ dht.diaglock.Lock()
396
+ if dht.IsListening(pmes.GetId()) {
397
+ //TODO: ehhh..........
398
+ dht.diaglock.Unlock()
399
+ return
400
+ }
401
+ dht.diaglock.Unlock()
402
+
403
+ seq := dht.routes[0].NearestPeers(convertPeerID(dht.self.ID), 10)
404
+ listen_chan := dht.ListenFor(pmes.GetId(), len(seq))
405
+
406
+ for _,ps := range seq {
407
+ mes := swarm.NewMessage(ps, pmes)
408
+ dht.network.Chan.Outgoing <-mes
409
+ }
410
+
411
+
412
+
413
+ buf := new(bytes.Buffer)
414
+ // NOTE: this shouldnt be a hardcoded value
415
+ after := time.After(time.Second * 20)
416
+ count := len(seq)
417
+ for count > 0 {
418
+ select {
419
+ case <-after:
420
+ //Timeout, return what we have
421
+ goto out
422
+ case req_resp := <-listen_chan:
423
+ buf.Write(req_resp.Data)
424
+ count--
425
+ }
426
+ }
427
+
428
+out:
429
+ di := dht.getDiagInfo()
430
+ buf.Write(di.Marshal())
431
+ resp := pDHTMessage{
432
+ Type: DHTMessage_DIAGNOSTIC,
433
+ Id: pmes.GetId(),
434
+ Value: buf.Bytes(),
435
+ Response: true,
436
+ }
437
+
438
+ mes := swarm.NewMessage(p, resp.ToProtobuf())
439
+ dht.network.Chan.Outgoing <-mes
440
+}
routing/dht/routing.go
+61
-10
@@ -3,6 +3,7 @@ package dht
3
import (
4
"math/rand"
5
"time"
6
+ "bytes"
7
"encoding/json"
8
9
proto "code.google.com/p/goprotobuf/proto"
@@ -30,7 +31,7 @@ func GenerateMessageID() uint64 {
31
// PutValue adds value corresponding to given Key.
32
func (s *IpfsDHT) PutValue(key u.Key, value []byte) error {
33
var p *peer.Peer
33
- p = s.routes.NearestPeer(convertKey(key))
34
+ p = s.routes[0].NearestPeer(convertKey(key))
35
if p == nil {
36
panic("Table returned nil peer!")
37
}
@@ -52,7 +53,7 @@ func (s *IpfsDHT) PutValue(key u.Key, value []byte) error {
53
// returned along with util.ErrSearchIncomplete
54
func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
55
var p *peer.Peer
55
- p = s.routes.NearestPeer(convertKey(key))
56
+ p = s.routes[0].NearestPeer(convertKey(key))
57
if p == nil {
58
panic("Table returned nil peer!")
59
}
@@ -62,7 +63,7 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
63
Key: string(key),
64
Id: GenerateMessageID(),
65
}
65
- response_chan := s.ListenFor(pmes.Id)
66
+ response_chan := s.ListenFor(pmes.Id, 1)
67
68
mes := swarm.NewMessage(p, pmes.ToProtobuf())
69
s.network.Chan.Outgoing <- mes
@@ -92,7 +93,7 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
93
94
// Announce that this node can provide value for given key
95
func (s *IpfsDHT) Provide(key u.Key) error {
95
- peers := s.routes.NearestPeers(convertKey(key), PoolSize)
96
+ peers := s.routes[0].NearestPeers(convertKey(key), PoolSize)
97
if len(peers) == 0 {
98
//return an error
99
}
@@ -112,7 +113,7 @@ func (s *IpfsDHT) Provide(key u.Key) error {
113
114
// FindProviders searches for peers who can provide the value for given key.
115
func (s *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer, error) {
115
- p := s.routes.NearestPeer(convertKey(key))
116
+ p := s.routes[0].NearestPeer(convertKey(key))
117
118
pmes := pDHTMessage{
119
Type: DHTMessage_GET_PROVIDERS,
@@ -122,7 +123,7 @@ func (s *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer,
123
124
mes := swarm.NewMessage(p, pmes.ToProtobuf())
125
125
- listen_chan := s.ListenFor(pmes.Id)
126
+ listen_chan := s.ListenFor(pmes.Id, 1)
127
u.DOut("Find providers for: '%s'", key)
128
s.network.Chan.Outgoing <-mes
129
after := time.After(timeout)
@@ -163,7 +164,6 @@ func (s *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer,
164
}
165
166
return prov_arr, nil
166
-
167
}
168
}
169
@@ -171,7 +171,7 @@ func (s *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer,
171
172
// FindPeer searches for a peer with given ID.
173
func (s *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, error) {
174
- p := s.routes.NearestPeer(convertPeerID(id))
174
+ p := s.routes[0].NearestPeer(convertPeerID(id))
175
176
pmes := pDHTMessage{
177
Type: DHTMessage_FIND_NODE,
@@ -181,7 +181,7 @@ func (s *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, error
181
182
mes := swarm.NewMessage(p, pmes.ToProtobuf())
183
184
- listen_chan := s.ListenFor(pmes.Id)
184
+ listen_chan := s.ListenFor(pmes.Id, 1)
185
s.network.Chan.Outgoing <-mes
186
after := time.After(timeout)
187
select {
@@ -224,7 +224,7 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
224
mes := swarm.NewMessage(p, pmes.ToProtobuf())
225
226
before := time.Now()
227
- response_chan := dht.ListenFor(pmes.Id)
227
+ response_chan := dht.ListenFor(pmes.Id, 1)
228
dht.network.Chan.Outgoing <- mes
229
230
tout := time.After(timeout)
@@ -241,3 +241,54 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
241
return u.ErrTimeout
242
}
243
}
244
+
245
+func (dht *IpfsDHT) GetDiagnostic(timeout time.Duration) ([]*diagInfo, error) {
246
+ u.DOut("Begin Diagnostic")
247
+ //Send to N closest peers
248
+ targets := dht.routes[0].NearestPeers(convertPeerID(dht.self.ID), 10)
249
+
250
+ // TODO: Add timeout to this struct so nodes know when to return
251
+ pmes := pDHTMessage{
252
+ Type: DHTMessage_DIAGNOSTIC,
253
+ Id: GenerateMessageID(),
254
+ }
255
+
256
+ listen_chan := dht.ListenFor(pmes.Id, len(targets))
257
+
258
+ pbmes := pmes.ToProtobuf()
259
+ for _,p := range targets {
260
+ mes := swarm.NewMessage(p, pbmes)
261
+ dht.network.Chan.Outgoing <-mes
262
+ }
263
+
264
+ var out []*diagInfo
265
+ after := time.After(timeout)
266
+ for count := len(targets); count > 0; {
267
+ select {
268
+ case <-after:
269
+ u.DOut("Diagnostic request timed out.")
270
+ return out, u.ErrTimeout
271
+ case resp := <-listen_chan:
272
+ pmes_out := new(DHTMessage)
273
+ err := proto.Unmarshal(resp.Data, pmes_out)
274
+ if err != nil {
275
+ // NOTE: here and elsewhere, need to audit error handling,
276
+ // some errors should be continued on from
277
+ return out, err
278
+ }
279
+
280
+ dec := json.NewDecoder(bytes.NewBuffer(pmes_out.GetValue()))
281
+ for {
282
+ di := new(diagInfo)
283
+ err := dec.Decode(di)
284
+ if err != nil {
285
+ break
286
+ }
287
+
288
+ out = append(out, di)
289
+ }
290
+ }
291
+ }
292
+
293
+ return nil,nil
294
+}
routing/dht/table.go
+2
-7
@@ -95,11 +95,7 @@ func (p peerSorterArr) Less(a, b int) bool {
95
//
96
97
func copyPeersFromList(target ID, peerArr peerSorterArr, peerList *list.List) peerSorterArr {
98
- if peerList == nil {
99
- return peerSorterArr{}
100
- }
101
- e := peerList.Front()
102
- for ; e != nil; {
98
+ for e := peerList.Front(); e != nil; e = e.Next() {
99
p := e.Value.(*peer.Peer)
100
p_id := convertPeerID(p.ID)
101
pd := peerDistance{
@@ -107,11 +103,10 @@ func copyPeersFromList(target ID, peerArr peerSorterArr, peerList *list.List) pe
103
distance: xor(target, p_id),
104
}
105
peerArr = append(peerArr, &pd)
110
- if e != nil {
106
+ if e == nil {
107
u.POut("list element was nil.")
108
return peerArr
109
}
114
- e = e.Next()
110
}
111
return peerArr
112
}