refactor routing to be more configurable. start extraction of routing to

new package.

Author: aschmahmann

dual/dual_test.go

@@ -15,6 +15,8 @@ import (
 	record "github.com/libp2p/go-libp2p-record"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
+
+	dhtrouting "github.com/libp2p/go-libp2p-kad-dht/routing"
 )
 
 var wancid, lancid cid.Cid
@@ -263,7 +265,7 @@ func TestSearchValue(t *testing.T) {
 
 	_ = wan.PutValue(ctx, "/v/hello", []byte("valid"))
 
-	valCh, err := d.SearchValue(ctx, "/v/hello", dht.Quorum(0))
+	valCh, err := d.SearchValue(ctx, "/v/hello", dhtrouting.Quorum(0))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -291,7 +293,7 @@ func TestSearchValue(t *testing.T) {
 		t.Error(err)
 	}
 
-	valCh, err = d.SearchValue(ctx, "/v/hello", dht.Quorum(0))
+	valCh, err = d.SearchValue(ctx, "/v/hello", dhtrouting.Quorum(0))
 	if err != nil {
 		t.Fatal(err)
 	}

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/multiformats/go-multiaddr-net v0.1.5
 	github.com/multiformats/go-multihash v0.0.13
 	github.com/multiformats/go-multistream v0.1.1
+	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.5.1
 	github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1
 	go.opencensus.io v0.22.3

go.sum

@@ -272,6 +272,7 @@ github.com/libp2p/go-reuseport v0.0.1/go.mod h1:jn6RmB1ufnQwl0Q1f+YxAj8isJgDCQza
 github.com/libp2p/go-reuseport-transport v0.0.2/go.mod h1:YkbSDrvjUVDL6b8XqriyA20obEtsW9BLkuOUyQAOCbs=
 github.com/libp2p/go-reuseport-transport v0.0.3 h1:zzOeXnTooCkRvoH+bSXEfXhn76+LAiwoneM0gnXjF2M=
 github.com/libp2p/go-reuseport-transport v0.0.3/go.mod h1:Spv+MPft1exxARzP2Sruj2Wb5JSyHNncjf1Oi2dEbzM=
+github.com/libp2p/go-sockaddr v0.0.2 h1:tCuXfpA9rq7llM/v834RKc/Xvovy/AqM9kHvTV/jY/Q=
 github.com/libp2p/go-sockaddr v0.0.2/go.mod h1:syPvOmNs24S3dFVGJA1/mrqdeijPxLV2Le3BRLKd68k=
 github.com/libp2p/go-stream-muxer v0.0.1/go.mod h1:bAo8x7YkSpadMTbtTaxGVHWUQsR/l5MEaHbKaliuT14=
 github.com/libp2p/go-stream-muxer-multistream v0.2.0 h1:714bRJ4Zy9mdhyTLJ+ZKiROmAFwUHpeRidG+q7LTQOg=

lookup.go

@@ -72,6 +72,11 @@ func (lk loggableKeyBytes) String() string {
 // If the context is canceled, this function will return the context error along
 // with the closest K peers it has found so far.
 func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) (<-chan peer.ID, error) {
+	return dht.GetClosestPeersSeeded(ctx, key, nil, true)
+}
+
+// GetClosestPeersSeeded is the Kademlia 'node lookup' operation
+func (dht *IpfsDHT) GetClosestPeersSeeded(ctx context.Context, key string, seedPeers []peer.ID, useRTPeers bool) (<-chan peer.ID, error) {
 	if key == "" {
 		return nil, fmt.Errorf("can't lookup empty key")
 	}
@@ -101,6 +106,7 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) (<-chan pee
 			return peers, err
 		},
 		func() bool { return false },
+		seedPeers, useRTPeers,
 	)
 
 	if err != nil {

query.go

@@ -77,9 +77,9 @@ type lookupWithFollowupResult struct {
 //
 // After the lookup is complete the query function is run (unless stopped) against all of the top K peers from the
 // lookup that have not already been successfully queried.
-func (dht *IpfsDHT) runLookupWithFollowup(ctx context.Context, target string, queryFn queryFn, stopFn stopFn) (*lookupWithFollowupResult, error) {
+func (dht *IpfsDHT) runLookupWithFollowup(ctx context.Context, target string, queryFn queryFn, stopFn stopFn, seedPeers []peer.ID, useRTPeers bool) (*lookupWithFollowupResult, error) {
 	// run the query
-	lookupRes, err := dht.runQuery(ctx, target, queryFn, stopFn)
+	lookupRes, err := dht.runQuery(ctx, target, queryFn, stopFn, seedPeers, useRTPeers)
 	if err != nil {
 		return nil, err
 	}
@@ -137,10 +137,23 @@ processFollowUp:
 	return lookupRes, nil
 }
 
-func (dht *IpfsDHT) runQuery(ctx context.Context, target string, queryFn queryFn, stopFn stopFn) (*lookupWithFollowupResult, error) {
+func (dht *IpfsDHT) runQuery(ctx context.Context, target string, queryFn queryFn, stopFn stopFn, manuallySeededPeers []peer.ID, useRTPeers bool) (*lookupWithFollowupResult, error) {
 	// pick the K closest peers to the key in our Routing table.
 	targetKadID := kb.ConvertKey(target)
-	seedPeers := dht.routingTable.NearestPeers(targetKadID, dht.bucketSize)
+
+	seedPeerSet := peer.NewSet()
+	for _, p := range manuallySeededPeers {
+		seedPeerSet.Add(p)
+	}
+
+	if manuallySeededPeers == nil || useRTPeers {
+		RTSeedPeers := dht.routingTable.NearestPeers(targetKadID, dht.bucketSize)
+		for _, p := range RTSeedPeers {
+			seedPeerSet.Add(p)
+		}
+	}
+
+	seedPeers := seedPeerSet.Peers()
 	if len(seedPeers) == 0 {
 		routing.PublishQueryEvent(ctx, &routing.QueryEvent{
 			Type:  routing.QueryError,