Asynchronous lookups

dht.go

@@ -92,6 +92,7 @@ type IpfsDHT struct {
 
 	bucketSize int
 	alpha      int // The concurrency parameter per path
+	beta       int // The number of peers closest to a target that must have responded for a query path to terminate
 	d          int // Number of Disjoint Paths to query
 
 	autoRefresh           bool
@@ -234,6 +235,7 @@ func makeDHT(ctx context.Context, h host.Host, cfg config) (*IpfsDHT, error) {
 		serverProtocols:   serverProtocols,
 		bucketSize:        cfg.bucketSize,
 		alpha:             cfg.concurrency,
+		beta:              cfg.resiliency,
 		d:                 cfg.disjointPaths,
 		triggerRtRefresh:  make(chan chan<- error),
 		triggerSelfLookup: make(chan chan<- error),

dht_bootstrap.go

@@ -8,7 +8,6 @@ import (
 	multierror "github.com/hashicorp/go-multierror"
 	process "github.com/jbenet/goprocess"
 	processctx "github.com/jbenet/goprocess/context"
-	"github.com/libp2p/go-libp2p-core/routing"
 	kbucket "github.com/libp2p/go-libp2p-kbucket"
 	"github.com/multiformats/go-multiaddr"
 	_ "github.com/multiformats/go-multiaddr-dns"
@@ -58,8 +57,8 @@ func (dht *IpfsDHT) startSelfLookup() error {
 
 			// Do a self walk
 			queryCtx, cancel := context.WithTimeout(ctx, dht.rtRefreshQueryTimeout)
-			_, err := dht.FindPeer(queryCtx, dht.self)
-			if err == routing.ErrNotFound || err == kbucket.ErrLookupFailure {
+			_, err := dht.GetClosestPeers(queryCtx, string(dht.self))
+			if err == kbucket.ErrLookupFailure {
 				err = nil
 			} else if err != nil {
 				err = fmt.Errorf("failed to query self during routing table refresh: %s", err)
@@ -207,10 +206,7 @@ func (dht *IpfsDHT) refreshCpls(ctx context.Context) error {
 
 		// walk to the generated peer
 		walkFnc := func(c context.Context) error {
-			_, err := dht.FindPeer(c, randPeer)
-			if err == routing.ErrNotFound {
-				return nil
-			}
+			_, err := dht.GetClosestPeers(c, string(randPeer))
 			return err
 		}
 

dht_options.go

@@ -34,6 +34,7 @@ type config struct {
 	bucketSize      int
 	disjointPaths   int
 	concurrency     int
+	resiliency      int
 	maxRecordAge    time.Duration
 	enableProviders bool
 	enableValues    bool
@@ -87,6 +88,7 @@ var defaults = func(o *config) error {
 
 	o.bucketSize = defaultBucketSize
 	o.concurrency = 3
+	o.resiliency = 3
 
 	return nil
 }
@@ -239,6 +241,17 @@ func Concurrency(alpha int) Option {
 	}
 }
 
+// Resiliency configures the number of peers closest to a target that must have responded in order for a given query
+// path to complete.
+//
+// The default value is 3.
+func Resiliency(beta int) Option {
+	return func(c *config) error {
+		c.resiliency = beta
+		return nil
+	}
+}
+
 // DisjointPaths configures the number of disjoint paths (d in the S/Kademlia paper) taken per query.
 //
 // The default value is BucketSize/2.

dht_test.go

@@ -1467,8 +1467,8 @@ func TestFindClosestPeers(t *testing.T) {
 		out = append(out, p)
 	}
 
-	if len(out) != querier.bucketSize {
-		t.Fatalf("got wrong number of peers (got %d, expected %d)", len(out), querier.bucketSize)
+	if len(out) < querier.beta {
+		t.Fatalf("got wrong number of peers (got %d, expected at least %d)", len(out), querier.beta)
 	}
 }