Merge pull request #456 from jbenet/provides-rewrite

rewrite of provides to better select peers to send RPCs to

core/mock.go

@@ -2,6 +2,7 @@ package core
 
 import (
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+
 	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
 	syncds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
 	"github.com/jbenet/go-ipfs/blocks/blockstore"

crypto/key.go

@@ -8,6 +8,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
+	"io"
 
 	"crypto/elliptic"
 	"crypto/hmac"
@@ -74,11 +75,15 @@ type PubKey interface {
 // Given a public key, generates the shared key.
 type GenSharedKey func([]byte) ([]byte, error)
 
-// Generates a keypair of the given type and bitsize
 func GenerateKeyPair(typ, bits int) (PrivKey, PubKey, error) {
+	return GenerateKeyPairWithReader(typ, bits, rand.Reader)
+}
+
+// Generates a keypair of the given type and bitsize
+func GenerateKeyPairWithReader(typ, bits int, src io.Reader) (PrivKey, PubKey, error) {
 	switch typ {
 	case RSA:
-		priv, err := rsa.GenerateKey(rand.Reader, bits)
+		priv, err := rsa.GenerateKey(src, bits)
 		if err != nil {
 			return nil, nil, err
 		}

crypto/key_test.go

@@ -1,12 +1,15 @@
-package crypto
+package crypto_test
 
 import (
+	. "github.com/jbenet/go-ipfs/crypto"
+
 	"bytes"
+	tu "github.com/jbenet/go-ipfs/util/testutil"
 	"testing"
 )
 
 func TestRsaKeys(t *testing.T) {
-	sk, pk, err := GenerateKeyPair(RSA, 512)
+	sk, pk, err := tu.RandKeyPair(512)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -90,7 +93,7 @@ func testKeyEquals(t *testing.T, k Key) {
 		t.Fatal("Key not equal to key with same bytes.")
 	}
 
-	sk, pk, err := GenerateKeyPair(RSA, 512)
+	sk, pk, err := tu.RandKeyPair(512)
 	if err != nil {
 		t.Fatal(err)
 	}

namesys/resolve_test.go

@@ -3,7 +3,6 @@ package namesys
 import (
 	"testing"
 
-	ci "github.com/jbenet/go-ipfs/crypto"
 	mockrouting "github.com/jbenet/go-ipfs/routing/mock"
 	u "github.com/jbenet/go-ipfs/util"
 	testutil "github.com/jbenet/go-ipfs/util/testutil"
@@ -15,7 +14,7 @@ func TestRoutingResolve(t *testing.T) {
 	resolver := NewRoutingResolver(d)
 	publisher := NewRoutingPublisher(d)
 
-	privk, pubk, err := ci.GenerateKeyPair(ci.RSA, 512)
+	privk, pubk, err := testutil.RandKeyPair(512)
 	if err != nil {
 		t.Fatal(err)
 	}

net/mock/mock_net.go

@@ -41,7 +41,7 @@ func New(ctx context.Context) Mocknet {
 }
 
 func (mn *mocknet) GenPeer() (inet.Network, error) {
-	sk, _, err := testutil.RandKeyPair(512)
+	sk, _, err := testutil.SeededKeyPair(int64(len(mn.nets)))
 	if err != nil {
 		return nil, err
 	}

peer/peer_test.go

@@ -1,4 +1,4 @@
-package peer
+package peer_test
 
 import (
 	"encoding/base64"
@@ -7,7 +7,9 @@ import (
 	"testing"
 
 	ic "github.com/jbenet/go-ipfs/crypto"
+	. "github.com/jbenet/go-ipfs/peer"
 	u "github.com/jbenet/go-ipfs/util"
+	tu "github.com/jbenet/go-ipfs/util/testutil"
 
 	b58 "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-base58"
 )
@@ -39,7 +41,7 @@ type keyset struct {
 
 func (ks *keyset) generate() error {
 	var err error
-	ks.sk, ks.pk, err = ic.GenerateKeyPair(ic.RSA, 1024)
+	ks.sk, ks.pk, err = tu.RandKeyPair(512)
 	if err != nil {
 		return err
 	}

routing/dht/dht.go

@@ -102,10 +102,9 @@ func (dht *IpfsDHT) Connect(ctx context.Context, npeer peer.ID) error {
 	return nil
 }
 
-// putValueToNetwork stores the given key/value pair at the peer 'p'
-// meaning: it sends a PUT_VALUE message to p
-func (dht *IpfsDHT) putValueToNetwork(ctx context.Context, p peer.ID,
-	key string, rec *pb.Record) error {
+// putValueToPeer stores the given key/value pair at the peer 'p'
+func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID,
+	key u.Key, rec *pb.Record) error {
 
 	pmes := pb.NewMessage(pb.Message_PUT_VALUE, string(key), 0)
 	pmes.Record = rec
@@ -238,12 +237,12 @@ func (dht *IpfsDHT) Update(ctx context.Context, p peer.ID) {
 }
 
 // FindLocal looks for a peer with a given ID connected to this dht and returns the peer and the table it was found in.
-func (dht *IpfsDHT) FindLocal(id peer.ID) (peer.PeerInfo, *kb.RoutingTable) {
+func (dht *IpfsDHT) FindLocal(id peer.ID) peer.PeerInfo {
 	p := dht.routingTable.Find(id)
 	if p != "" {
-		return dht.peerstore.PeerInfo(p), dht.routingTable
+		return dht.peerstore.PeerInfo(p)
 	}
-	return peer.PeerInfo{}, nil
+	return peer.PeerInfo{}
 }
 
 // findPeerSingle asks peer 'p' if they know where the peer with id 'id' is
@@ -257,26 +256,6 @@ func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key u.Ke
 	return dht.sendRequest(ctx, p, pmes)
 }
 
-func (dht *IpfsDHT) addProviders(key u.Key, pbps []*pb.Message_Peer) []peer.ID {
-	peers := pb.PBPeersToPeerInfos(pbps)
-
-	var provArr []peer.ID
-	for _, pi := range peers {
-		p := pi.ID
-
-		// Dont add outselves to the list
-		if p == dht.self {
-			continue
-		}
-
-		log.Debugf("%s adding provider: %s for %s", dht.self, p, key)
-		// TODO(jbenet) ensure providers is idempotent
-		dht.providers.AddProvider(key, p)
-		provArr = append(provArr, p)
-	}
-	return provArr
-}
-
 // nearestPeersToQuery returns the routing tables closest peers.
 func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
 	key := u.Key(pmes.GetKey())
@@ -285,7 +264,7 @@ func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
 }
 
 // betterPeerToQuery returns nearestPeersToQuery, but iff closer than self.
-func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, count int) []peer.ID {
+func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, p peer.ID, count int) []peer.ID {
 	closer := dht.nearestPeersToQuery(pmes, count)
 
 	// no node? nil
@@ -302,11 +281,16 @@ func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, count int) []peer.ID {
 	}
 
 	var filtered []peer.ID
-	for _, p := range closer {
+	for _, clp := range closer {
+		// Dont send a peer back themselves
+		if p == clp {
+			continue
+		}
+
 		// must all be closer than self
 		key := u.Key(pmes.GetKey())
-		if !kb.Closer(dht.self, p, key) {
-			filtered = append(filtered, p)
+		if !kb.Closer(dht.self, clp, key) {
+			filtered = append(filtered, clp)
 		}
 	}
 
@@ -323,23 +307,6 @@ func (dht *IpfsDHT) ensureConnectedToPeer(ctx context.Context, p peer.ID) error
 	return dht.network.DialPeer(ctx, p)
 }
 
-//TODO: this should be smarter about which keys it selects.
-func (dht *IpfsDHT) loadProvidableKeys() error {
-	kl, err := dht.datastore.KeyList()
-	if err != nil {
-		return err
-	}
-	for _, dsk := range kl {
-		k := u.KeyFromDsKey(dsk)
-		if len(k) == 0 {
-			log.Errorf("loadProvidableKeys error: %v", dsk)
-		}
-
-		dht.providers.AddProvider(k, dht.self)
-	}
-	return nil
-}
-
 // PingRoutine periodically pings nearest neighbors.
 func (dht *IpfsDHT) PingRoutine(t time.Duration) {
 	defer dht.Children().Done()

routing/dht/dht_test.go

@@ -14,7 +14,6 @@ import (
 	dssync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
 	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 
-	// ci "github.com/jbenet/go-ipfs/crypto"
 	inet "github.com/jbenet/go-ipfs/net"
 	peer "github.com/jbenet/go-ipfs/peer"
 	routing "github.com/jbenet/go-ipfs/routing"
@@ -35,7 +34,7 @@ func init() {
 
 func setupDHT(ctx context.Context, t *testing.T, addr ma.Multiaddr) *IpfsDHT {
 
-	sk, pk, err := testutil.RandKeyPair(512)
+	sk, pk, err := testutil.SeededKeyPair(time.Now().UnixNano())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -487,12 +486,7 @@ func TestLayeredGet(t *testing.T) {
 	connect(t, ctx, dhts[1], dhts[2])
 	connect(t, ctx, dhts[1], dhts[3])
 
-	err := dhts[3].putLocal(u.Key("/v/hello"), []byte("world"))
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = dhts[3].Provide(ctx, u.Key("/v/hello"))
+	err := dhts[3].Provide(ctx, u.Key("/v/hello"))
 	if err != nil {
 		t.Fatal(err)
 	}

routing/dht/ext_test.go

@@ -47,9 +47,8 @@ func TestGetFailures(t *testing.T) {
 		t.Fatal("Did not get expected error!")
 	}
 
-	msgs := make(chan *pb.Message, 100)
+	t.Log("Timeout test passed.")
 
-	// u.POut("NotFound Test\n")
 	// Reply with failures to every message
 	nets[1].SetHandler(inet.ProtocolDHT, func(s inet.Stream) {
 		defer s.Close()
@@ -68,8 +67,6 @@ func TestGetFailures(t *testing.T) {
 		if err := pbw.WriteMsg(resp); err != nil {
 			panic(err)
 		}
-
-		msgs <- resp
 	})
 
 	// This one should fail with NotFound
@@ -83,6 +80,8 @@ func TestGetFailures(t *testing.T) {
 		t.Fatal("expected error, got none.")
 	}
 
+	t.Log("ErrNotFound check passed!")
+
 	// Now we test this DHT's handleGetValue failure
 	{
 		typ := pb.Message_GET_VALUE

routing/dht/handlers.go

@@ -39,7 +39,7 @@ func (dht *IpfsDHT) handlerForMsgType(t pb.Message_MessageType) dhtHandler {
 }
 
 func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
-	log.Debugf("%s handleGetValue for key: %s\n", dht.self, pmes.GetKey())
+	log.Debugf("%s handleGetValue for key: %s", dht.self, pmes.GetKey())
 
 	// setup response
 	resp := pb.NewMessage(pmes.GetType(), pmes.GetKey(), pmes.GetClusterLevel())
@@ -93,7 +93,7 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 	}
 
 	// Find closest peer on given cluster to desired key and reply with that info
-	closer := dht.betterPeersToQuery(pmes, CloserPeerCount)
+	closer := dht.betterPeersToQuery(pmes, p, CloserPeerCount)
 	closerinfos := peer.PeerInfos(dht.peerstore, closer)
 	if closer != nil {
 		for _, pi := range closerinfos {
@@ -127,7 +127,7 @@ func (dht *IpfsDHT) handlePutValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 	}
 
 	err = dht.datastore.Put(dskey, data)
-	log.Debugf("%s handlePutValue %v\n", dht.self, dskey)
+	log.Debugf("%s handlePutValue %v", dht.self, dskey)
 	return pmes, err
 }
 
@@ -144,11 +144,11 @@ func (dht *IpfsDHT) handleFindPeer(ctx context.Context, p peer.ID, pmes *pb.Mess
 	if peer.ID(pmes.GetKey()) == dht.self {
 		closest = []peer.ID{dht.self}
 	} else {
-		closest = dht.betterPeersToQuery(pmes, CloserPeerCount)
+		closest = dht.betterPeersToQuery(pmes, p, CloserPeerCount)
 	}
 
 	if closest == nil {
-		log.Debugf("handleFindPeer: could not find anything.")
+		log.Warningf("handleFindPeer: could not find anything.")
 		return resp, nil
 	}
 
@@ -189,7 +189,7 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.ID, pmes *pb.
 	}
 
 	// Also send closer peers.
-	closer := dht.betterPeersToQuery(pmes, CloserPeerCount)
+	closer := dht.betterPeersToQuery(pmes, p, CloserPeerCount)
 	if closer != nil {
 		infos := peer.PeerInfos(dht.peerstore, providers)
 		resp.CloserPeers = pb.PeerInfosToPBPeers(dht.network, infos)