refactor peer distance search + handleGetProviders
Juan Batiz-Benet committed
Sep 16, 2014 at 01:54 UTC
69ed45c555db599591211d2003dee1533368ee29
1 file changed
+62
-51
routing/dht/dht.go
+62
-51
@@ -245,24 +245,8 @@ func (dht *IpfsDHT) handleGetValue(p *peer.Peer, pmes *Message) (*Message, error
245
}
246
247
// Find closest peer on given cluster to desired key and reply with that info
248
- // TODO: this should probably be decomposed.
249
-
250
- // stored levels are > 1, to distinguish missing levels.
251
- level := pmes.GetClusterLevel()
252
- u.DOut("handleGetValue searching level %d clusters\n", level)
253
-
254
- ck := kb.ConvertKey(u.Key(pmes.GetKey()))
255
- closer := dht.routingTables[level].NearestPeer(ck)
256
-
257
- // if closer peer is self, return nil
258
- if closer.ID.Equal(dht.self.ID) {
259
- u.DOut("Attempted to return self! this shouldnt happen...\n")
260
- resp.CloserPeers = nil
261
- return resp, nil
262
- }
263
-
264
- // if self is closer than the one from the table, return nil
265
- if kb.Closer(dht.self.ID, closer.ID, u.Key(pmes.GetKey())) {
248
+ closer := dht.betterPeerToQuery(pmes)
249
+ if closer == nil {
250
u.DOut("handleGetValue could not find a closer node than myself.\n")
251
resp.CloserPeers = nil
252
return resp, nil
@@ -293,12 +277,15 @@ func (dht *IpfsDHT) handlePing(p *peer.Peer, pmes *Message) (*Message, error) {
277
278
func (dht *IpfsDHT) handleFindPeer(p *peer.Peer, pmes *Message) (*Message, error) {
279
resp := &Message{Type: pmes.Type}
280
+ var closest *peer.Peer
281
297
- level := pmes.GetClusterLevel()
298
- u.DOut("handleFindPeer: searching for '%s'\n", peer.ID(pmes.GetKey()).Pretty())
282
+ // if looking for self... special case where we send it on CloserPeers.
283
+ if peer.ID(pmes.GetKey()).Equal(dht.self.ID) {
284
+ closest = dht.self
285
+ } else {
286
+ closest = dht.betterPeerToQuery(pmes)
287
+ }
288
300
- ck := kb.ConvertKey(u.Key(pmes.GetKey()))
301
- closest := dht.routingTables[level].NearestPeer(ck)
289
if closest == nil {
290
u.PErr("handleFindPeer: could not find anything.\n")
291
return resp, nil
@@ -309,52 +296,42 @@ func (dht *IpfsDHT) handleFindPeer(p *peer.Peer, pmes *Message) (*Message, error
296
return resp, nil
297
}
298
312
- // If the found peer further away than this peer...
313
- if kb.Closer(dht.self.ID, closest.ID, u.Key(pmes.GetKey())) {
314
- return resp, nil
315
- }
316
-
299
u.DOut("handleFindPeer: sending back '%s'\n", closest.ID.Pretty())
300
resp.CloserPeers = peersToPBPeers([]*peer.Peer{closest})
301
return resp, nil
302
}
303
322
-func (dht *IpfsDHT) handleGetProviders(p *peer.Peer, pmes *Message) {
323
- resp := Message{
324
- Type: Message_GET_PROVIDERS,
325
- Key: pmes.GetKey(),
326
- ID: pmes.GetId(),
327
- Response: true,
304
+func (dht *IpfsDHT) handleGetProviders(p *peer.Peer, pmes *Message) (*Message, error) {
305
+ resp := &Message{
306
+ Type: pmes.Type,
307
+ Key: pmes.Key,
308
}
309
310
+ // check if we have this value, to add ourselves as provider.
311
has, err := dht.datastore.Has(ds.NewKey(pmes.GetKey()))
331
- if err != nil {
332
- dht.netChan.Errors <- err
312
+ if err != nil && err != ds.ErrNotFound {
313
+ u.PErr("unexpected datastore error: %v\n", err)
314
+ has = false
315
}
316
317
+ // setup providers
318
providers := dht.providers.GetProviders(u.Key(pmes.GetKey()))
319
if has {
320
providers = append(providers, dht.self)
321
}
339
- if providers == nil || len(providers) == 0 {
340
- level := 0
341
- if len(pmes.GetValue()) > 0 {
342
- level = int(pmes.GetValue()[0])
343
- }
322
345
- closer := dht.routingTables[level].NearestPeer(kb.ConvertKey(u.Key(pmes.GetKey())))
346
- if kb.Closer(dht.self.ID, closer.ID, u.Key(pmes.GetKey())) {
347
- resp.Peers = nil
348
- } else {
349
- resp.Peers = []*peer.Peer{closer}
350
- }
351
- } else {
352
- resp.Peers = providers
353
- resp.Success = true
323
+ // if we've got providers, send thos those.
324
+ if providers != nil && len(providers) > 0 {
325
+ resp.ProviderPeers = peersToPBPeers(providers)
326
}
327
356
- mes := swarm.NewMessage(p, resp.ToProtobuf())
357
- dht.netChan.Outgoing <- mes
328
+ // Also send closer peers.
329
+ closer := dht.betterPeerToQuery(pmes)
330
+ if closer != nil {
331
+ resp.CloserPeers = peersToPBPeers([]*peer.Peer{closer})
332
+ }
333
+
334
+ return resp, nil
335
}
336
337
type providerInfo struct {
@@ -671,6 +648,40 @@ func (dht *IpfsDHT) addPeerList(key u.Key, peers []*Message_PBPeer) []*peer.Peer
648
return provArr
649
}
650
651
+// nearestPeerToQuery returns the routing tables closest peers.
652
+func (dht *IpfsDHT) nearestPeerToQuery(pmes *Message) *peer.Peer {
653
+ level := pmes.GetClusterLevel()
654
+ cluster := dht.routingTables[level]
655
+
656
+ key := u.Key(pmes.GetKey())
657
+ closer := cluster.NearestPeer(kb.ConvertKey(key))
658
+ return closer
659
+}
660
+
661
+// betterPeerToQuery returns nearestPeerToQuery, but iff closer than self.
662
+func (dht *IpfsDHT) betterPeerToQuery(pmes *Message) *peer.Peer {
663
+ closer := dht.nearestPeerToQuery(pmes)
664
+
665
+ // no node? nil
666
+ if closer == nil {
667
+ return nil
668
+ }
669
+
670
+ // == to self? nil
671
+ if closer.ID.Equal(dht.self.ID) {
672
+ u.DOut("Attempted to return self! this shouldnt happen...\n")
673
+ return nil
674
+ }
675
+
676
+ // self is closer? nil
677
+ if kb.Closer(dht.self.ID, closer.ID, key) {
678
+ return nil
679
+ }
680
+
681
+ // ok seems like a closer node.
682
+ return closer
683
+}
684
+
685
func (dht *IpfsDHT) peerFromInfo(pbp *Message_PBPeer) (*peer.Peer, error) {
686
maddr, err := ma.NewMultiaddr(pbp.GetAddr())
687
if err != nil {