mirror of
https://github.com/fluencelabs/go-libp2p-kad-dht
synced 2025-04-24 22:32:13 +00:00
Routing uses context now
@perfmode boom
This commit is contained in:
Juan Batiz-Benet
Brian Tiger Chow
parent
0a9810a150
commit
a62411035e
8
dht.go
8
dht.go
@ -73,7 +73,7 @@ func NewDHT(p *peer.Peer, ps peer.Peerstore, net inet.Network, sender inet.Sende
|
||||
}
|
||||
|
||||
// Connect to a new peer at the given address, ping and add to the routing table
|
||||
func (dht *IpfsDHT) Connect(npeer *peer.Peer) (*peer.Peer, error) {
|
||||
func (dht *IpfsDHT) Connect(ctx context.Context, npeer *peer.Peer) (*peer.Peer, error) {
|
||||
u.DOut("Connect to new peer: %s\n", npeer.ID.Pretty())
|
||||
|
||||
// TODO(jbenet,whyrusleeping)
|
||||
@ -92,7 +92,7 @@ func (dht *IpfsDHT) Connect(npeer *peer.Peer) (*peer.Peer, error) {
|
||||
|
||||
// Ping new peer to register in their routing table
|
||||
// NOTE: this should be done better...
|
||||
err = dht.Ping(npeer, time.Second*2)
|
||||
err = dht.Ping(ctx, npeer)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to ping newly connected peer: %s\n", err)
|
||||
}
|
||||
@ -497,8 +497,8 @@ func (dht *IpfsDHT) loadProvidableKeys() error {
|
||||
}
|
||||
|
||||
// Bootstrap builds up list of peers by requesting random peer IDs
|
||||
func (dht *IpfsDHT) Bootstrap() {
|
||||
func (dht *IpfsDHT) Bootstrap(ctx context.Context) {
|
||||
id := make([]byte, 16)
|
||||
rand.Read(id)
|
||||
dht.FindPeer(peer.ID(id), time.Second*10)
|
||||
dht.FindPeer(ctx, peer.ID(id))
|
||||
}
|
||||
|
||||
74
dht_test.go
74
dht_test.go
@ -22,7 +22,7 @@ import (
|
||||
)
|
||||
|
||||
func setupDHT(t *testing.T, p *peer.Peer) *IpfsDHT {
|
||||
ctx, _ := context.WithCancel(context.TODO())
|
||||
ctx := context.Background()
|
||||
|
||||
peerstore := peer.NewPeerstore()
|
||||
|
||||
@ -150,9 +150,11 @@ func TestValueGetSet(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
dhtA.PutValue("hello", []byte("world"))
|
||||
ctxT, _ := context.WithTimeout(context.Background(), time.Second)
|
||||
dhtA.PutValue(ctxT, "hello", []byte("world"))
|
||||
|
||||
val, err := dhtA.GetValue("hello", time.Second*2)
|
||||
ctxT, _ = context.WithTimeout(context.Background(), time.Second*2)
|
||||
val, err := dhtA.GetValue(ctxT, "hello")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -208,7 +210,8 @@ func TestProvides(t *testing.T) {
|
||||
|
||||
time.Sleep(time.Millisecond * 60)
|
||||
|
||||
provs, err := dhts[0].FindProviders(u.Key("hello"), time.Second)
|
||||
ctxT, _ := context.WithTimeout(context.Background(), time.Second)
|
||||
provs, err := dhts[0].FindProviders(ctxT, u.Key("hello"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -218,6 +221,63 @@ func TestProvides(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvidesAsync(t *testing.T) {
|
||||
// t.Skip("skipping test to debug another")
|
||||
|
||||
u.Debug = false
|
||||
|
||||
_, peers, dhts := setupDHTS(4, t)
|
||||
defer func() {
|
||||
for i := 0; i < 4; i++ {
|
||||
dhts[i].Halt()
|
||||
defer dhts[i].network.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
_, err := dhts[0].Connect(peers[1])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_, err = dhts[1].Connect(peers[2])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_, err = dhts[1].Connect(peers[3])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = dhts[3].putLocal(u.Key("hello"), []byte("world"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
bits, err := dhts[3].getLocal(u.Key("hello"))
|
||||
if err != nil && bytes.Equal(bits, []byte("world")) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = dhts[3].Provide(u.Key("hello"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond * 60)
|
||||
|
||||
ctx, _ := context.WithTimeout(context.TODO(), time.Millisecond*300)
|
||||
provs := dhts[0].FindProvidersAsync(ctx, u.Key("hello"), 5)
|
||||
select {
|
||||
case p := <-provs:
|
||||
if !p.ID.Equal(dhts[3].self.ID) {
|
||||
t.Fatalf("got a provider, but not the right one. %v", p.ID.Pretty())
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatal("Didnt get back providers")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLayeredGet(t *testing.T) {
|
||||
// t.Skip("skipping test to debug another")
|
||||
|
||||
@ -257,7 +317,8 @@ func TestLayeredGet(t *testing.T) {
|
||||
|
||||
time.Sleep(time.Millisecond * 60)
|
||||
|
||||
val, err := dhts[0].GetValue(u.Key("hello"), time.Second)
|
||||
ctxT, _ := context.WithTimeout(context.Background(), time.Second)
|
||||
val, err := dhts[0].GetValue(ctxT, u.Key("hello"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -296,7 +357,8 @@ func TestFindPeer(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
p, err := dhts[0].FindPeer(peers[2].ID, time.Second)
|
||||
ctxT, _ := context.WithTimeout(context.Background(), time.Second)
|
||||
p, err := dhts[0].FindPeer(ctxT, peers[2].ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
12
ext_test.go
12
ext_test.go
@ -108,7 +108,8 @@ func TestGetFailures(t *testing.T) {
|
||||
|
||||
// This one should time out
|
||||
// u.POut("Timout Test\n")
|
||||
_, err := d.GetValue(u.Key("test"), time.Millisecond*10)
|
||||
ctx1, _ := context.WithTimeout(context.Background(), time.Second)
|
||||
_, err := d.GetValue(ctx1, u.Key("test"))
|
||||
if err != nil {
|
||||
if err != context.DeadlineExceeded {
|
||||
t.Fatal("Got different error than we expected", err)
|
||||
@ -134,7 +135,8 @@ func TestGetFailures(t *testing.T) {
|
||||
})
|
||||
|
||||
// This one should fail with NotFound
|
||||
_, err = d.GetValue(u.Key("test"), time.Millisecond*1000)
|
||||
ctx2, _ := context.WithTimeout(context.Background(), time.Second)
|
||||
_, err = d.GetValue(ctx2, u.Key("test"))
|
||||
if err != nil {
|
||||
if err != u.ErrNotFound {
|
||||
t.Fatalf("Expected ErrNotFound, got: %s", err)
|
||||
@ -236,7 +238,8 @@ func TestNotFound(t *testing.T) {
|
||||
|
||||
})
|
||||
|
||||
v, err := d.GetValue(u.Key("hello"), time.Second*5)
|
||||
ctx, _ := context.WithTimeout(context.Background(), time.Second*5)
|
||||
v, err := d.GetValue(ctx, u.Key("hello"))
|
||||
u.DOut("get value got %v\n", v)
|
||||
if err != nil {
|
||||
switch err {
|
||||
@ -299,7 +302,8 @@ func TestLessThanKResponses(t *testing.T) {
|
||||
|
||||
})
|
||||
|
||||
_, err := d.GetValue(u.Key("hello"), time.Second*30)
|
||||
ctx, _ := context.WithTimeout(context.Background(), time.Second*30)
|
||||
_, err := d.GetValue(ctx, u.Key("hello"))
|
||||
if err != nil {
|
||||
switch err {
|
||||
case u.ErrNotFound:
|
||||
|
||||
29
routing.go
29
routing.go
@ -3,7 +3,6 @@ package dht
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
|
||||
|
||||
@ -18,9 +17,7 @@ import (
|
||||
|
||||
// PutValue adds value corresponding to given Key.
|
||||
// This is the top level "Store" operation of the DHT
|
||||
func (dht *IpfsDHT) PutValue(key u.Key, value []byte) error {
|
||||
ctx := context.TODO()
|
||||
|
||||
func (dht *IpfsDHT) PutValue(ctx context.Context, key u.Key, value []byte) error {
|
||||
peers := []*peer.Peer{}
|
||||
|
||||
// get the peers we need to announce to
|
||||
@ -46,12 +43,10 @@ func (dht *IpfsDHT) PutValue(key u.Key, value []byte) error {
|
||||
// GetValue searches for the value corresponding to given Key.
|
||||
// If the search does not succeed, a multiaddr string of a closer peer is
|
||||
// returned along with util.ErrSearchIncomplete
|
||||
func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
|
||||
func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
|
||||
ll := startNewRPC("GET")
|
||||
defer ll.EndAndPrint()
|
||||
|
||||
ctx, _ := context.WithTimeout(context.TODO(), timeout)
|
||||
|
||||
// If we have it local, dont bother doing an RPC!
|
||||
// NOTE: this might not be what we want to do...
|
||||
val, err := dht.getLocal(key)
|
||||
@ -101,8 +96,7 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
|
||||
// This is what DSHTs (Coral and MainlineDHT) do to store large values in a DHT.
|
||||
|
||||
// Provide makes this node announce that it can provide a value for the given key
|
||||
func (dht *IpfsDHT) Provide(key u.Key) error {
|
||||
ctx := context.TODO()
|
||||
func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
|
||||
|
||||
dht.providers.AddProvider(key, dht.self)
|
||||
peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), PoolSize)
|
||||
@ -174,12 +168,10 @@ func (dht *IpfsDHT) addPeerListAsync(k u.Key, peers []*Message_Peer, ps *peerSet
|
||||
}
|
||||
|
||||
// FindProviders searches for peers who can provide the value for given key.
|
||||
func (dht *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Peer, error) {
|
||||
func (dht *IpfsDHT) FindProviders(ctx context.Context, key u.Key) ([]*peer.Peer, error) {
|
||||
ll := startNewRPC("FindProviders")
|
||||
ll.EndAndPrint()
|
||||
|
||||
ctx, _ := context.WithTimeout(context.TODO(), timeout)
|
||||
|
||||
// get closest peer
|
||||
u.DOut("Find providers for: '%s'\n", key)
|
||||
p := dht.routingTables[0].NearestPeer(kb.ConvertKey(key))
|
||||
@ -223,8 +215,7 @@ func (dht *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Pee
|
||||
// Find specific Peer
|
||||
|
||||
// FindPeer searches for a peer with given ID.
|
||||
func (dht *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, error) {
|
||||
ctx, _ := context.WithTimeout(context.TODO(), timeout)
|
||||
func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (*peer.Peer, error) {
|
||||
|
||||
// Check if were already connected to them
|
||||
p, _ := dht.Find(id)
|
||||
@ -266,8 +257,7 @@ func (dht *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, err
|
||||
return nil, u.ErrNotFound
|
||||
}
|
||||
|
||||
func (dht *IpfsDHT) findPeerMultiple(id peer.ID, timeout time.Duration) (*peer.Peer, error) {
|
||||
ctx, _ := context.WithTimeout(context.TODO(), timeout)
|
||||
func (dht *IpfsDHT) findPeerMultiple(ctx context.Context, id peer.ID) (*peer.Peer, error) {
|
||||
|
||||
// Check if were already connected to them
|
||||
p, _ := dht.Find(id)
|
||||
@ -325,9 +315,7 @@ func (dht *IpfsDHT) findPeerMultiple(id peer.ID, timeout time.Duration) (*peer.P
|
||||
}
|
||||
|
||||
// Ping a peer, log the time it took
|
||||
func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
|
||||
ctx, _ := context.WithTimeout(context.TODO(), timeout)
|
||||
|
||||
func (dht *IpfsDHT) Ping(ctx context.Context, p *peer.Peer) error {
|
||||
// Thoughts: maybe this should accept an ID and do a peer lookup?
|
||||
u.DOut("Enter Ping.\n")
|
||||
|
||||
@ -336,8 +324,7 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (dht *IpfsDHT) getDiagnostic(timeout time.Duration) ([]*diagInfo, error) {
|
||||
ctx, _ := context.WithTimeout(context.TODO(), timeout)
|
||||
func (dht *IpfsDHT) getDiagnostic(ctx context.Context) ([]*diagInfo, error) {
|
||||
|
||||
u.DOut("Begin Diagnostic")
|
||||
peers := dht.routingTables[0].NearestPeers(kb.ConvertPeerID(dht.self.ID), 10)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user