Fix dht queries
Queries previously would sometimes only query three (alpha value) peers before halting the operation. This PR changes the number of peers grabbed from the routing table to start a query to K. Dht nodes would also not respond with enough peers, as per the kademlia paper, this has been changed to from 4 to 'K'. The query mechanism itself also was flawed in that it would pull all the peers it had yet to query out of the queue and 'start' the query for them. The concurrency rate limiting was done inside the 'queryPeer' method after the goroutine was spawned. This did not allow for peers receiver from query replies to be properly queried in order of distance. License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Sep 21, 2015 at 09:55 UTC
235a9ec5fcbb2ba6cc9099e3d77ef386ab62398b
5 files changed
+23
-28
routing/dht/dht.go
+1
-5
@@ -312,11 +312,7 @@ func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, p peer.ID, count int) [
312
continue
313
}
314
315
- // must all be closer than self
316
- key := key.Key(pmes.GetKey())
317
- if !kb.Closer(dht.self, clp, key) {
318
- filtered = append(filtered, clp)
319
- }
315
+ filtered = append(filtered, clp)
316
}
317
318
// ok seems like closer nodes
routing/dht/handlers.go
+1
-1
@@ -14,7 +14,7 @@ import (
14
)
15
16
// The number of closer peers to send on requests.
17
-var CloserPeerCount = 4
17
+var CloserPeerCount = KValue
18
19
// dhthandler specifies the signature of functions that handle DHT messages.
20
type dhtHandler func(context.Context, peer.ID, *pb.Message) (*pb.Message, error)
routing/dht/lookup.go
+1
-1
@@ -23,7 +23,7 @@ func pointerizePeerInfos(pis []peer.PeerInfo) []*peer.PeerInfo {
23
// to the given key
24
func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key key.Key) (<-chan peer.ID, error) {
25
e := log.EventBegin(ctx, "getClosestPeers", &key)
26
- tablepeers := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue)
26
+ tablepeers := dht.routingTable.NearestPeers(kb.ConvertKey(key), KValue)
27
if len(tablepeers) == 0 {
28
return nil, kb.ErrLookupFailure
29
}
routing/dht/query.go
+16
-17
@@ -184,29 +184,28 @@ func (r *dhtQueryRunner) spawnWorkers(proc process.Process) {
184
case <-r.proc.Closing():
185
return
186
187
- case p, more := <-r.peersToQuery.DeqChan:
188
- if !more {
189
- return // channel closed.
187
+ case <-r.rateLimit:
188
+ select {
189
+ case p, more := <-r.peersToQuery.DeqChan:
190
+ if !more {
191
+ return // channel closed.
192
+ }
193
+
194
+ // do it as a child func to make sure Run exits
195
+ // ONLY AFTER spawn workers has exited.
196
+ proc.Go(func(proc process.Process) {
197
+ r.queryPeer(proc, p)
198
+ })
199
+ case <-r.proc.Closing():
200
+ return
201
+ case <-r.peersRemaining.Done():
202
+ return
203
}
191
-
192
- // do it as a child func to make sure Run exits
193
- // ONLY AFTER spawn workers has exited.
194
- proc.Go(func(proc process.Process) {
195
- r.queryPeer(proc, p)
196
- })
204
}
205
}
206
}
207
208
func (r *dhtQueryRunner) queryPeer(proc process.Process, p peer.ID) {
202
- // make sure we rate limit concurrency.
203
- select {
204
- case <-r.rateLimit:
205
- case <-proc.Closing():
206
- r.peersRemaining.Decrement(1)
207
- return
208
- }
209
-
209
// ok let's do this!
210
211
// create a context from our proc.
routing/dht/routing.go
+4
-4
@@ -145,7 +145,7 @@ func (dht *IpfsDHT) GetValues(ctx context.Context, key key.Key, nvals int) ([]ro
145
}
146
147
// get closest peers in the routing table
148
- rtp := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue)
148
+ rtp := dht.routingTable.NearestPeers(kb.ConvertKey(key), KValue)
149
log.Debugf("peers in rt: %s", len(rtp), rtp)
150
if len(rtp) == 0 {
151
log.Warning("No peers from routing table!")
@@ -322,7 +322,7 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key key.Key,
322
return &dhtQueryResult{closerPeers: clpeers}, nil
323
})
324
325
- peers := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue)
325
+ peers := dht.routingTable.NearestPeers(kb.ConvertKey(key), KValue)
326
_, err := query.Run(ctx, peers)
327
if err != nil {
328
log.Debugf("Query error: %s", err)
@@ -342,7 +342,7 @@ func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (peer.PeerInfo, er
342
return pi, nil
343
}
344
345
- peers := dht.routingTable.NearestPeers(kb.ConvertPeerID(id), AlphaValue)
345
+ peers := dht.routingTable.NearestPeers(kb.ConvertPeerID(id), KValue)
346
if len(peers) == 0 {
347
return peer.PeerInfo{}, kb.ErrLookupFailure
348
}
@@ -409,7 +409,7 @@ func (dht *IpfsDHT) FindPeersConnectedToPeer(ctx context.Context, id peer.ID) (<
409
peerchan := make(chan peer.PeerInfo, asyncQueryBuffer)
410
peersSeen := peer.Set{}
411
412
- peers := dht.routingTable.NearestPeers(kb.ConvertPeerID(id), AlphaValue)
412
+ peers := dht.routingTable.NearestPeers(kb.ConvertPeerID(id), KValue)
413
if len(peers) == 0 {
414
return nil, kb.ErrLookupFailure
415
}