wip with DHT
@whyrusleeping @jbenet this is a WIP with the DHT. wip License: MIT Signed-off-by: Brian Tiger Chow <brian@perfmode.com> Conflicts: epictest/addcat_test.go exchange/bitswap/testnet/peernet.go exchange/bitswap/testutils.go routing/mock/centralized_server.go routing/mock/centralized_test.go routing/mock/interface.go fix(routing/mock) fill in function definition
Brian Tiger Chow committed
Dec 17, 2014 at 10:02 UTC
ca32a83394a19311eae8661cd6400d00ed1bd578
15 files changed
+126
-67
core/mock.go
+5
-4
@@ -1,6 +1,7 @@
1
package core
2
3
import (
4
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
6
syncds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
7
"github.com/jbenet/go-ipfs/blocks/blockstore"
@@ -11,12 +12,13 @@ import (
12
nsys "github.com/jbenet/go-ipfs/namesys"
13
path "github.com/jbenet/go-ipfs/path"
14
peer "github.com/jbenet/go-ipfs/peer"
14
- mdht "github.com/jbenet/go-ipfs/routing/mock"
15
+ dht "github.com/jbenet/go-ipfs/routing/dht"
16
ds2 "github.com/jbenet/go-ipfs/util/datastore2"
17
)
18
19
// NewMockNode constructs an IpfsNode for use in tests.
20
func NewMockNode() (*IpfsNode, error) {
21
+ ctx := context.TODO()
22
nd := new(IpfsNode)
23
24
// Generate Identity
@@ -41,8 +43,7 @@ func NewMockNode() (*IpfsNode, error) {
43
nd.Datastore = ds2.CloserWrap(syncds.MutexWrap(dstore))
44
45
// Routing
44
- dht := mdht.NewServer().ClientWithDatastore(peer.PeerInfo{ID: p}, nd.Datastore)
45
- nd.Routing = dht
46
+ nd.Routing = dht.NewDHT(ctx, nd.Identity, nd.Network, nd.Datastore)
47
48
// Bitswap
49
bstore := blockstore.NewBlockstore(nd.Datastore)
@@ -54,7 +55,7 @@ func NewMockNode() (*IpfsNode, error) {
55
nd.DAG = mdag.NewDAGService(bserv)
56
57
// Namespace resolver
57
- nd.Namesys = nsys.NewNameSystem(dht)
58
+ nd.Namesys = nsys.NewNameSystem(nd.Routing)
59
60
// Path resolver
61
nd.Resolver = &path.Resolver{DAG: nd.DAG}
epictest/addcat_test.go
+31
-5
@@ -4,6 +4,7 @@ import (
4
"bytes"
5
"fmt"
6
"io"
7
+ "math"
8
"os"
9
"testing"
10
"time"
@@ -16,12 +17,12 @@ import (
17
importer "github.com/jbenet/go-ipfs/importer"
18
chunk "github.com/jbenet/go-ipfs/importer/chunk"
19
merkledag "github.com/jbenet/go-ipfs/merkledag"
20
+ mocknet "github.com/jbenet/go-ipfs/net/mock"
21
path "github.com/jbenet/go-ipfs/path"
22
mockrouting "github.com/jbenet/go-ipfs/routing/mock"
23
uio "github.com/jbenet/go-ipfs/unixfs/io"
24
util "github.com/jbenet/go-ipfs/util"
25
errors "github.com/jbenet/go-ipfs/util/debugerror"
24
- delay "github.com/jbenet/go-ipfs/util/delay"
26
)
27
28
const kSeed = 1
@@ -87,11 +88,14 @@ func RandomBytes(n int64) []byte {
88
89
func AddCatBytes(data []byte, conf Config) error {
90
ctx := context.Background()
90
- rs := mockrouting.NewServerWithDelay(mockrouting.DelayConfig{
91
- Query: delay.Fixed(conf.RoutingLatency),
92
- ValueVisibility: delay.Fixed(conf.RoutingLatency),
91
+ mn := mocknet.New(ctx)
92
+ // defer mn.Close() FIXME does mocknet require clean-up
93
+ mn.SetLinkDefaults(mocknet.LinkOptions{
94
+ Latency: conf.NetworkLatency,
95
+ Bandwidth: math.MaxInt32, // TODO add to conf
96
})
94
- net, err := tn.StreamNetWithDelay(ctx, rs, delay.Fixed(conf.NetworkLatency))
97
+ dhtNetwork := mockrouting.NewDHTNetwork(mn)
98
+ net, err := tn.StreamNet(ctx, mn, dhtNetwork)
99
if err != nil {
100
return errors.Wrap(err)
101
}
@@ -100,6 +104,28 @@ func AddCatBytes(data []byte, conf Config) error {
104
105
adder := sessionGenerator.Next()
106
catter := sessionGenerator.Next()
107
+ // catter.Routing.Update(context.TODO(), adder.Peer)
108
+
109
+ peers := mn.Peers()
110
+ if len(peers) != 2 {
111
+ return errors.New("peers not in network")
112
+ }
113
+
114
+ for _, i := range peers {
115
+ for _, j := range peers {
116
+ if i == j {
117
+ continue
118
+ }
119
+ fmt.Println(i, " and ", j)
120
+ if _, err := mn.LinkPeers(i, j); err != nil {
121
+ return err
122
+ }
123
+ if err := mn.ConnectPeers(i, j); err != nil {
124
+ return err
125
+ }
126
+ }
127
+ }
128
+
129
catter.SetBlockstoreLatency(conf.BlockstoreLatency)
130
131
adder.SetBlockstoreLatency(0) // disable blockstore latency during add operation
epictest/bench_test.go
+1
@@ -18,6 +18,7 @@ func benchmarkAddCat(numBytes int64, conf Config, b *testing.B) {
18
19
var instant = Config{}.All_Instantaneous()
20
21
+func BenchmarkInstantaneousAddCat1KB(b *testing.B) { benchmarkAddCat(1*KB, instant, b) }
22
func BenchmarkInstantaneousAddCat1MB(b *testing.B) { benchmarkAddCat(1*MB, instant, b) }
23
func BenchmarkInstantaneousAddCat2MB(b *testing.B) { benchmarkAddCat(2*MB, instant, b) }
24
func BenchmarkInstantaneousAddCat4MB(b *testing.B) { benchmarkAddCat(4*MB, instant, b) }
exchange/bitswap/bitswap_test.go
+2
-2
@@ -11,10 +11,10 @@ import (
11
blocks "github.com/jbenet/go-ipfs/blocks"
12
blocksutil "github.com/jbenet/go-ipfs/blocks/blocksutil"
13
tn "github.com/jbenet/go-ipfs/exchange/bitswap/testnet"
14
- peer "github.com/jbenet/go-ipfs/peer"
14
mockrouting "github.com/jbenet/go-ipfs/routing/mock"
15
u "github.com/jbenet/go-ipfs/util"
16
delay "github.com/jbenet/go-ipfs/util/delay"
17
+ "github.com/jbenet/go-ipfs/util/testutil"
18
)
19
20
// FIXME the tests are really sensitive to the network delay. fix them to work
@@ -61,7 +61,7 @@ func TestProviderForKeyButNetworkCannotFind(t *testing.T) { // TODO revisit this
61
defer g.Close()
62
63
block := blocks.NewBlock([]byte("block"))
64
- pinfo := peer.PeerInfo{ID: peer.ID("testing")}
64
+ pinfo := testutil.RandPeerOrFatal(t)
65
rs.Client(pinfo).Provide(context.Background(), block.Key()) // but not on network
66
67
solo := g.Next()
exchange/bitswap/testnet/peernet.go
+3
-14
@@ -1,14 +1,12 @@
1
package bitswap
2
3
import (
4
- "math"
5
-
4
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
+ ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
6
bsnet "github.com/jbenet/go-ipfs/exchange/bitswap/network"
7
mockpeernet "github.com/jbenet/go-ipfs/net/mock"
8
peer "github.com/jbenet/go-ipfs/peer"
9
mockrouting "github.com/jbenet/go-ipfs/routing/mock"
11
- delay "github.com/jbenet/go-ipfs/util/delay"
10
testutil "github.com/jbenet/go-ipfs/util/testutil"
11
)
12
@@ -17,16 +15,7 @@ type peernet struct {
15
routingserver mockrouting.Server
16
}
17
20
-func StreamNetWithDelay(
21
- ctx context.Context,
22
- rs mockrouting.Server,
23
- d delay.D) (Network, error) {
24
-
25
- net := mockpeernet.New(ctx)
26
- net.SetLinkDefaults(mockpeernet.LinkOptions{
27
- Latency: d.Get(),
28
- Bandwidth: math.MaxInt32, // TODO inject
29
- })
18
+func StreamNet(ctx context.Context, net mockpeernet.Mocknet, rs mockrouting.Server) (Network, error) {
19
return &peernet{net, rs}, nil
20
}
21
@@ -39,7 +28,7 @@ func (pn *peernet) Adapter(p testutil.Peer) bsnet.BitSwapNetwork {
28
for _, other := range peers {
29
pn.Mocknet.LinkPeers(p.ID(), other)
30
}
42
- routing := pn.routingserver.Client(peer.PeerInfo{ID: p.ID()})
31
+ routing := pn.routingserver.ClientWithDatastore(context.TODO(), p, ds.NewMapDatastore())
32
return bsnet.NewFromIpfsNetwork(client, routing)
33
}
34
exchange/bitswap/testnet/virtual.go
+1
-1
@@ -33,7 +33,7 @@ func (n *network) Adapter(p testutil.Peer) bsnet.BitSwapNetwork {
33
client := &networkClient{
34
local: p.ID(),
35
network: n,
36
- routing: n.routingserver.Client(peer.PeerInfo{ID: p.ID()}),
36
+ routing: n.routingserver.Client(p),
37
}
38
n.clients[p.ID()] = client
39
return client
exchange/bitswap/testutils.go
+5
-5
@@ -79,15 +79,15 @@ func (i *Instance) SetBlockstoreLatency(t time.Duration) time.Duration {
79
// sessions. To safeguard, use the SessionGenerator to generate sessions. It's
80
// just a much better idea.
81
func session(ctx context.Context, net tn.Network, p testutil.Peer) Instance {
82
+ bsdelay := delay.Fixed(0)
83
+ const kWriteCacheElems = 100
84
85
adapter := net.Adapter(p)
86
+ dstore := ds_sync.MutexWrap(datastore2.WithDelay(ds.NewMapDatastore(), bsdelay))
87
85
- bsdelay := delay.Fixed(0)
86
- const kWriteCacheElems = 100
87
- bstore, err := blockstore.WriteCached(blockstore.NewBlockstore(ds_sync.MutexWrap(datastore2.WithDelay(ds.NewMapDatastore(), bsdelay))), kWriteCacheElems)
88
+ bstore, err := blockstore.WriteCached(blockstore.NewBlockstore(ds_sync.MutexWrap(dstore)), kWriteCacheElems)
89
if err != nil {
89
- // FIXME perhaps change signature and return error.
90
- panic(err.Error())
90
+ panic(err.Error()) // FIXME perhaps change signature and return error.
91
}
92
93
const alwaysSendToPeer = true
namesys/resolve_test.go
+1
-6
@@ -4,18 +4,13 @@ import (
4
"testing"
5
6
ci "github.com/jbenet/go-ipfs/crypto"
7
- peer "github.com/jbenet/go-ipfs/peer"
7
mockrouting "github.com/jbenet/go-ipfs/routing/mock"
8
u "github.com/jbenet/go-ipfs/util"
9
testutil "github.com/jbenet/go-ipfs/util/testutil"
10
)
11
12
func TestRoutingResolve(t *testing.T) {
14
- local, err := testutil.RandPeerID()
15
- if err != nil {
16
- t.Fatal(err)
17
- }
18
- d := mockrouting.NewServer().Client(peer.PeerInfo{ID: local})
13
+ d := mockrouting.NewServer().Client(testutil.RandPeerOrFatal(t))
14
15
resolver := NewRoutingResolver(d)
16
publisher := NewRoutingPublisher(d)
net/mux.go
+1
@@ -76,6 +76,7 @@ func (m *Mux) ReadProtocolHeader(s io.Reader) (string, StreamHandler, error) {
76
// This operation is threadsafe.
77
func (m *Mux) SetHandler(p ProtocolID, h StreamHandler) {
78
m.Lock()
79
+ log.Debug("setting protocol ", p)
80
m.Handlers[p] = h
81
m.Unlock()
82
}
routing/dht/routing.go
+10
@@ -1,6 +1,7 @@
1
package dht
2
3
import (
4
+ "math"
5
"sync"
6
7
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -127,6 +128,15 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
128
return nil
129
}
130
131
+// FindProviders searches until the context expires.
132
+func (dht *IpfsDHT) FindProviders(ctx context.Context, key u.Key) ([]peer.PeerInfo, error) {
133
+ var providers []peer.PeerInfo
134
+ for p := range dht.FindProvidersAsync(ctx, key, math.MaxInt32) {
135
+ providers = append(providers, p)
136
+ }
137
+ return providers, nil
138
+}
139
+
140
// FindProvidersAsync is the same thing as FindProviders, but returns a channel.
141
// Peers will be returned on the channel as soon as they are found, even before
142
// the search query completes.
routing/mock/centralized_client.go
+8
-2
@@ -5,9 +5,11 @@ import (
5
6
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
7
ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
8
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
9
peer "github.com/jbenet/go-ipfs/peer"
10
routing "github.com/jbenet/go-ipfs/routing"
11
u "github.com/jbenet/go-ipfs/util"
12
+ "github.com/jbenet/go-ipfs/util/testutil"
13
)
14
15
var log = u.Logger("mockrouter")
@@ -15,7 +17,7 @@ var log = u.Logger("mockrouter")
17
type client struct {
18
datastore ds.Datastore
19
server server
18
- peer peer.PeerInfo
20
+ peer testutil.Peer
21
}
22
23
// FIXME(brian): is this method meant to simulate putting a value into the network?
@@ -70,7 +72,11 @@ func (c *client) FindProvidersAsync(ctx context.Context, k u.Key, max int) <-cha
72
// Provide returns once the message is on the network. Value is not necessarily
73
// visible yet.
74
func (c *client) Provide(_ context.Context, key u.Key) error {
73
- return c.server.Announce(c.peer, key)
75
+ info := peer.PeerInfo{
76
+ ID: c.peer.ID(),
77
+ Addrs: []ma.Multiaddr{c.peer.Address()},
78
+ }
79
+ return c.server.Announce(info, key)
80
}
81
82
var _ routing.IpfsRouting = &client{}
routing/mock/centralized_server.go
+5
-3
@@ -5,9 +5,11 @@ import (
5
"sync"
6
"time"
7
8
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
10
peer "github.com/jbenet/go-ipfs/peer"
11
u "github.com/jbenet/go-ipfs/util"
12
+ "github.com/jbenet/go-ipfs/util/testutil"
13
)
14
15
// server is the mockrouting.Client's private interface to the routing server
@@ -71,11 +73,11 @@ func (rs *s) Providers(k u.Key) []peer.PeerInfo {
73
return ret
74
}
75
74
-func (rs *s) Client(p peer.PeerInfo) Client {
75
- return rs.ClientWithDatastore(p, ds.NewMapDatastore())
76
+func (rs *s) Client(p testutil.Peer) Client {
77
+ return rs.ClientWithDatastore(context.Background(), p, ds.NewMapDatastore())
78
}
79
78
-func (rs *s) ClientWithDatastore(p peer.PeerInfo, datastore ds.Datastore) Client {
80
+func (rs *s) ClientWithDatastore(_ context.Context, p testutil.Peer, datastore ds.Datastore) Client {
81
return &client{
82
peer: p,
83
datastore: ds.NewMapDatastore(),
routing/mock/centralized_test.go
+12
-22
@@ -8,11 +8,12 @@ import (
8
peer "github.com/jbenet/go-ipfs/peer"
9
u "github.com/jbenet/go-ipfs/util"
10
delay "github.com/jbenet/go-ipfs/util/delay"
11
+ "github.com/jbenet/go-ipfs/util/testutil"
12
)
13
14
func TestKeyNotFound(t *testing.T) {
15
15
- var pi = peer.PeerInfo{ID: peer.ID("the peer id")}
16
+ var pi = testutil.RandPeerOrFatal(t)
17
var key = u.Key("mock key")
18
var ctx = context.Background()
19
@@ -25,7 +26,7 @@ func TestKeyNotFound(t *testing.T) {
26
}
27
28
func TestClientFindProviders(t *testing.T) {
28
- pi := peer.PeerInfo{ID: peer.ID("42")}
29
+ pi := testutil.RandPeerOrFatal(t)
30
rs := NewServer()
31
client := rs.Client(pi)
32
@@ -39,20 +40,6 @@ func TestClientFindProviders(t *testing.T) {
40
time.Sleep(time.Millisecond * 300)
41
max := 100
42
42
- providersFromHashTable, err := rs.Client(pi).FindProviders(context.Background(), k)
43
- if err != nil {
44
- t.Fatal(err)
45
- }
46
-
47
- isInHT := false
48
- for _, pi := range providersFromHashTable {
49
- if pi.ID == pi.ID {
50
- isInHT = true
51
- }
52
- }
53
- if !isInHT {
54
- t.Fatal("Despite client providing key, peer wasn't in hash table as a provider")
55
- }
43
providersFromClient := client.FindProvidersAsync(context.Background(), u.Key("hello"), max)
44
isInClient := false
45
for pi := range providersFromClient {
@@ -70,7 +57,7 @@ func TestClientOverMax(t *testing.T) {
57
k := u.Key("hello")
58
numProvidersForHelloKey := 100
59
for i := 0; i < numProvidersForHelloKey; i++ {
73
- pi := peer.PeerInfo{ID: peer.ID(i)}
60
+ pi := testutil.RandPeerOrFatal(t)
61
err := rs.Client(pi).Provide(context.Background(), k)
62
if err != nil {
63
t.Fatal(err)
@@ -78,7 +65,7 @@ func TestClientOverMax(t *testing.T) {
65
}
66
67
max := 10
81
- pi := peer.PeerInfo{ID: peer.ID("TODO")}
68
+ pi := testutil.RandPeerOrFatal(t)
69
client := rs.Client(pi)
70
71
providersFromClient := client.FindProvidersAsync(context.Background(), k, max)
@@ -113,8 +100,11 @@ func TestCanceledContext(t *testing.T) {
100
default:
101
}
102
116
- pi := peer.PeerInfo{ID: peer.ID(i)}
117
- err := rs.Client(pi).Provide(context.Background(), k)
103
+ pi, err := testutil.RandPeer()
104
+ if err != nil {
105
+ t.Error(err)
106
+ }
107
+ err = rs.Client(pi).Provide(context.Background(), k)
108
if err != nil {
109
t.Error(err)
110
}
@@ -122,7 +112,7 @@ func TestCanceledContext(t *testing.T) {
112
}
113
}()
114
125
- local := peer.PeerInfo{ID: peer.ID("peer id doesn't matter")}
115
+ local := testutil.RandPeerOrFatal(t)
116
client := rs.Client(local)
117
118
t.Log("warning: max is finite so this test is non-deterministic")
@@ -148,7 +138,7 @@ func TestCanceledContext(t *testing.T) {
138
139
func TestValidAfter(t *testing.T) {
140
151
- var pi = peer.PeerInfo{ID: peer.ID("the peer id")}
141
+ pi := testutil.RandPeerOrFatal(t)
142
var key = u.Key("mock key")
143
var ctx = context.Background()
144
conf := DelayConfig{
routing/mock/interface.go
+3
-3
@@ -11,18 +11,18 @@ import (
11
routing "github.com/jbenet/go-ipfs/routing"
12
u "github.com/jbenet/go-ipfs/util"
13
delay "github.com/jbenet/go-ipfs/util/delay"
14
+ "github.com/jbenet/go-ipfs/util/testutil"
15
)
16
17
// Server provides mockrouting Clients
18
type Server interface {
18
- Client(p peer.PeerInfo) Client
19
- ClientWithDatastore(peer.PeerInfo, ds.Datastore) Client
19
+ Client(p testutil.Peer) Client
20
+ ClientWithDatastore(context.Context, testutil.Peer, ds.Datastore) Client
21
}
22
23
// Client implements IpfsRouting
24
type Client interface {
25
FindProviders(context.Context, u.Key) ([]peer.PeerInfo, error)
25
-
26
routing.IpfsRouting
27
}
28
routing/mock/server2.go
new
+38
@@ -0,0 +1,38 @@
1
+package mockrouting
2
+
3
+import (
4
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
+ ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
6
+ sync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
7
+ mocknet "github.com/jbenet/go-ipfs/net/mock"
8
+ dht "github.com/jbenet/go-ipfs/routing/dht"
9
+ "github.com/jbenet/go-ipfs/util/testutil"
10
+)
11
+
12
+type mocknetserver struct {
13
+ mn mocknet.Mocknet
14
+}
15
+
16
+func NewDHTNetwork(mn mocknet.Mocknet) Server {
17
+ return &mocknetserver{
18
+ mn: mn,
19
+ }
20
+}
21
+
22
+func (rs *mocknetserver) Client(p testutil.Peer) Client {
23
+ return rs.ClientWithDatastore(context.TODO(), p, ds.NewMapDatastore())
24
+}
25
+
26
+func (rs *mocknetserver) ClientWithDatastore(ctx context.Context, p testutil.Peer, ds ds.Datastore) Client {
27
+
28
+ // FIXME AddPeer doesn't appear to be idempotent
29
+
30
+ net, err := rs.mn.AddPeer(p.PrivateKey(), p.Address())
31
+ if err != nil {
32
+ panic("FIXME")
33
+ // return nil, debugerror.Wrap(err)
34
+ }
35
+ return dht.NewDHT(ctx, p.ID(), net, sync.MutexWrap(ds))
36
+}
37
+
38
+var _ Server = &mocknetserver{}