change Provide RPC to not wait for an ACK, improves performance of 'Add' operations

dht/dht.go

@@ -120,15 +120,12 @@ func (dht *IpfsDHT) putProvider(ctx context.Context, p peer.Peer, key string) er
 	// add self as the provider
 	pmes.ProviderPeers = pb.PeersToPBPeers(dht.network, []peer.Peer{dht.self})
 
-	rpmes, err := dht.sendRequest(ctx, p, pmes)
+	err := dht.sendMessage(ctx, p, pmes)
 	if err != nil {
 		return err
 	}
 
 	log.Debugf("%s putProvider: %s for %s", dht.self, p, u.Key(key))
-	if rpmes.GetKey() != pmes.GetKey() {
-		return errors.New("provider not added correctly")
-	}
 
 	return nil
 }

dht/dht_net.go

@@ -8,8 +8,8 @@ import (
 	peer "github.com/jbenet/go-ipfs/peer"
 	pb "github.com/jbenet/go-ipfs/routing/dht/pb"
 
-	ggio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	ggio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
 )
 
 // handleNewStream implements the inet.StreamHandler
@@ -102,3 +102,24 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.Peer, pmes *pb.Messa
 	log.Event(ctx, "dhtReceivedMessage", dht.self, p, rpmes)
 	return rpmes, nil
 }
+
+// sendMessage sends out a message
+func (dht *IpfsDHT) sendMessage(ctx context.Context, p peer.Peer, pmes *pb.Message) error {
+
+	log.Debugf("%s dht starting stream", dht.self)
+	s, err := dht.network.NewStream(inet.ProtocolDHT, p)
+	if err != nil {
+		return err
+	}
+	defer s.Close()
+
+	w := ggio.NewDelimitedWriter(s)
+
+	log.Debugf("%s writing", dht.self)
+	if err := w.WriteMsg(pmes); err != nil {
+		return err
+	}
+	log.Event(ctx, "dhtSentMessage", dht.self, p, pmes)
+	log.Debugf("%s done", dht.self)
+	return nil
+}

mock/mockrouting_test.go

@@ -36,6 +36,9 @@ func TestClientFindProviders(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+
+	// This is bad... but simulating networks is hard
+	time.Sleep(time.Millisecond * 300)
 	max := 100
 
 	providersFromHashTable, err := rs.Client(peer).FindProviders(context.Background(), k)
@@ -160,6 +163,7 @@ func TestValidAfter(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	t.Log("providers", providers)
 	if len(providers) != 1 {
 		t.Fail()
 	}