coreapi: dht: simplify the implementation
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Jul 19, 2018 at 13:18 UTC
906c747064f155ffa24568c835989c8c3b89990c
3 files changed
+35
-155
core/coreapi/dht.go
+15
-131
@@ -9,143 +9,47 @@ import (
9
caopts "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
10
11
dag "gx/ipfs/QmNr4E8z9bGTztvHJktp7uQaMdx9p3r9Asrq6eYk7iCh4a/go-merkledag"
12
+ offline "gx/ipfs/QmPuLWvxK1vg6ckKUpT53Dow9VLCcQGdL5Trwxa8PTLp7r/go-ipfs-exchange-offline"
13
+ blockservice "gx/ipfs/QmQLG22wSEStiociTSKQpZAuuaaWoF1B3iKyjPFvWiTQ77/go-blockservice"
14
peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
13
- ipdht "gx/ipfs/QmRNxiPpZf3skMAtmDJpgHuW9uj1ukqV1zjANj9d6bmHfE/go-libp2p-kad-dht"
14
- ipld "gx/ipfs/QmX5CsuHyVZeTLxgRSYkgLSDQKb9UjE8xnhQzCEJWWWFsC/go-ipld-format"
15
routing "gx/ipfs/QmY9JUvS8kbgao3XbPh6WAV3ChE2nxGKhcGTHiwMC4gmcU/go-libp2p-routing"
16
- notif "gx/ipfs/QmY9JUvS8kbgao3XbPh6WAV3ChE2nxGKhcGTHiwMC4gmcU/go-libp2p-routing/notifications"
17
- ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"
16
cid "gx/ipfs/QmZFbDTY9jfSBms2MchvYM9oYRbAF19K7Pby47yDBfpPrb/go-cid"
17
pstore "gx/ipfs/Qmda4cPRvSRyox3SqgJN6DfSZGU5TtHufPTp9uXjFj71X6/go-libp2p-peerstore"
18
+ blockstore "gx/ipfs/Qmeg56ecxRnVv7VWViMrDeEMoBHaNFMs4vQnyQrJ79Zz7i/go-ipfs-blockstore"
19
)
20
22
-var ErrNotDHT = errors.New("routing service is not a DHT")
23
-
21
type DhtAPI struct {
22
*CoreAPI
23
*caopts.DhtOptions
24
}
25
29
-func (api *DhtAPI) FindPeer(ctx context.Context, p peer.ID) (<-chan ma.Multiaddr, error) {
30
- dht, ok := api.node.Routing.(*ipdht.IpfsDHT)
31
- if !ok {
32
- return nil, ErrNotDHT
26
+func (api *DhtAPI) FindPeer(ctx context.Context, p peer.ID) (pstore.PeerInfo, error) {
27
+ pi, err := api.node.Routing.FindPeer(ctx, peer.ID(p))
28
+ if err != nil {
29
+ return pstore.PeerInfo{}, err
30
}
31
35
- outChan := make(chan ma.Multiaddr)
36
- events := make(chan *notif.QueryEvent)
37
- ctx = notif.RegisterForQueryEvents(ctx, events)
38
-
39
- go func() {
40
- defer close(outChan)
41
-
42
- sendAddrs := func(responses []*pstore.PeerInfo) error {
43
- for _, response := range responses {
44
- for _, addr := range response.Addrs {
45
- select {
46
- case outChan <- addr:
47
- case <-ctx.Done():
48
- return ctx.Err()
49
- }
50
- }
51
- }
52
- return nil
53
- }
54
-
55
- for event := range events {
56
- if event.Type == notif.FinalPeer {
57
- err := sendAddrs(event.Responses)
58
- if err != nil {
59
- return
60
- }
61
- }
62
- }
63
- }()
64
-
65
- go func() {
66
- defer close(events)
67
- pi, err := dht.FindPeer(ctx, peer.ID(p))
68
- if err != nil {
69
- notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
70
- Type: notif.QueryError,
71
- Extra: err.Error(),
72
- })
73
- return
74
- }
75
-
76
- notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
77
- Type: notif.FinalPeer,
78
- Responses: []*pstore.PeerInfo{&pi},
79
- })
80
- }()
81
-
82
- return outChan, nil
32
+ return pi, nil
33
}
34
85
-func (api *DhtAPI) FindProviders(ctx context.Context, p coreiface.Path, opts ...caopts.DhtFindProvidersOption) (<-chan peer.ID, error) {
35
+func (api *DhtAPI) FindProviders(ctx context.Context, p coreiface.Path, opts ...caopts.DhtFindProvidersOption) (<-chan pstore.PeerInfo, error) {
36
settings, err := caopts.DhtFindProvidersOptions(opts...)
37
if err != nil {
38
return nil, err
39
}
40
91
- dht, ok := api.node.Routing.(*ipdht.IpfsDHT)
92
- if !ok {
93
- return nil, ErrNotDHT
94
- }
95
-
41
rp, err := api.ResolvePath(ctx, p)
42
if err != nil {
43
return nil, err
44
}
45
101
- c := rp.Cid()
102
-
46
numProviders := settings.NumProviders
47
if numProviders < 1 {
48
return nil, fmt.Errorf("number of providers must be greater than 0")
49
}
50
108
- outChan := make(chan peer.ID)
109
- events := make(chan *notif.QueryEvent)
110
- ctx = notif.RegisterForQueryEvents(ctx, events)
111
-
112
- pchan := dht.FindProvidersAsync(ctx, c, numProviders)
113
- go func() {
114
- defer close(outChan)
115
-
116
- sendProviders := func(responses []*pstore.PeerInfo) error {
117
- for _, response := range responses {
118
- select {
119
- case outChan <- response.ID:
120
- case <-ctx.Done():
121
- return ctx.Err()
122
- }
123
- }
124
- return nil
125
- }
126
-
127
- for event := range events {
128
- if event.Type == notif.Provider {
129
- err := sendProviders(event.Responses)
130
- if err != nil {
131
- return
132
- }
133
- }
134
- }
135
- }()
136
-
137
- go func() {
138
- defer close(events)
139
- for p := range pchan {
140
- np := p
141
- notif.PublishQueryEvent(ctx, ¬if.QueryEvent{
142
- Type: notif.Provider,
143
- Responses: []*pstore.PeerInfo{&np},
144
- })
145
- }
146
- }()
147
-
148
- return outChan, nil
51
+ pchan := api.node.Routing.FindProvidersAsync(ctx, rp.Cid(), numProviders)
52
+ return pchan, nil
53
}
54
55
func (api *DhtAPI) Provide(ctx context.Context, path coreiface.Path, opts ...caopts.DhtProvideOption) error {
@@ -158,10 +62,6 @@ func (api *DhtAPI) Provide(ctx context.Context, path coreiface.Path, opts ...cao
62
return errors.New("cannot provide in offline mode")
63
}
64
161
- if len(api.node.PeerHost.Network().Conns()) == 0 {
162
- return errors.New("cannot provide, no connected peers")
163
- }
164
-
65
rp, err := api.ResolvePath(ctx, path)
66
if err != nil {
67
return err
@@ -178,26 +78,8 @@ func (api *DhtAPI) Provide(ctx context.Context, path coreiface.Path, opts ...cao
78
return fmt.Errorf("block %s not found locally, cannot provide", c)
79
}
80
181
- //TODO: either remove or use
182
- //outChan := make(chan interface{})
183
-
184
- //events := make(chan *notif.QueryEvent)
185
- //ctx = notif.RegisterForQueryEvents(ctx, events)
186
-
187
- /*go func() {
188
- defer close(outChan)
189
- for range events {
190
- select {
191
- case <-ctx.Done():
192
- return
193
- default:
194
- }
195
- }
196
- }()*/
197
-
198
- //defer close(events)
81
if settings.Recursive {
200
- err = provideKeysRec(ctx, api.node.Routing, api.node.DAG, []*cid.Cid{c})
82
+ err = provideKeysRec(ctx, api.node.Routing, api.node.Blockstore, []*cid.Cid{c})
83
} else {
84
err = provideKeys(ctx, api.node.Routing, []*cid.Cid{c})
85
}
@@ -218,11 +100,13 @@ func provideKeys(ctx context.Context, r routing.IpfsRouting, cids []*cid.Cid) er
100
return nil
101
}
102
221
-func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv ipld.DAGService, cids []*cid.Cid) error {
103
+func provideKeysRec(ctx context.Context, r routing.IpfsRouting, bs blockstore.Blockstore, cids []*cid.Cid) error {
104
provided := cid.NewSet()
105
for _, c := range cids {
106
kset := cid.NewSet()
107
108
+ dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
109
+
110
err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, kset.Visit)
111
if err != nil {
112
return err
core/coreapi/dht_test.go
+16
-20
@@ -6,10 +6,10 @@ import (
6
"io/ioutil"
7
"testing"
8
9
- coreapi "github.com/ipfs/go-ipfs/core/coreapi"
9
+ "github.com/ipfs/go-ipfs/core/coreapi/interface"
10
11
- blocks "gx/ipfs/QmR54CzE4UcdFAZDehj6HFyy3eSHhVsJUpjfnhCmscuStS/go-block-format"
12
- peer "gx/ipfs/QmdVrMn1LhB4ybb8hMVaMLXnA8XRSewMnK6YqXKXoTcRvN/go-libp2p-peer"
11
+ peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
12
+ blocks "gx/ipfs/QmWAzSEoqZ6xU6pu8yL8e5WaMb7wtbfbhhN4p1DknUPtr3/go-block-format"
13
)
14
15
func TestDhtFindPeer(t *testing.T) {
@@ -19,26 +19,22 @@ func TestDhtFindPeer(t *testing.T) {
19
t.Fatal(err)
20
}
21
22
- out, err := apis[2].Dht().FindPeer(ctx, peer.ID(nds[0].Identity))
22
+ pi, err := apis[2].Dht().FindPeer(ctx, peer.ID(nds[0].Identity))
23
if err != nil {
24
t.Fatal(err)
25
}
26
27
- addr := <-out
28
-
29
- if addr.String() != "/ip4/127.0.0.1/tcp/4001" {
30
- t.Errorf("got unexpected address from FindPeer: %s", addr.String())
27
+ if pi.Addrs[0].String() != "/ip4/127.0.0.1/tcp/4001" {
28
+ t.Errorf("got unexpected address from FindPeer: %s", pi.Addrs[0].String())
29
}
30
33
- out, err = apis[1].Dht().FindPeer(ctx, peer.ID(nds[2].Identity))
31
+ pi, err = apis[1].Dht().FindPeer(ctx, peer.ID(nds[2].Identity))
32
if err != nil {
33
t.Fatal(err)
34
}
35
38
- addr = <-out
39
-
40
- if addr.String() != "/ip4/127.0.2.1/tcp/4001" {
41
- t.Errorf("got unexpected address from FindPeer: %s", addr.String())
36
+ if pi.Addrs[0].String() != "/ip4/127.0.2.1/tcp/4001" {
37
+ t.Errorf("got unexpected address from FindPeer: %s", pi.Addrs[0].String())
38
}
39
}
40
@@ -61,8 +57,8 @@ func TestDhtFindProviders(t *testing.T) {
57
58
provider := <-out
59
64
- if provider.String() != nds[0].Identity.String() {
65
- t.Errorf("got wrong provider: %s != %s", provider.String(), nds[0].Identity.String())
60
+ if provider.ID.String() != nds[0].Identity.String() {
61
+ t.Errorf("got wrong provider: %s != %s", provider.ID.String(), nds[0].Identity.String())
62
}
63
}
64
@@ -81,7 +77,7 @@ func TestDhtProvide(t *testing.T) {
77
78
b := blocks.NewBlock(data)
79
nds[0].Blockstore.Put(b)
84
- p := coreapi.ParseCid(b.Cid())
80
+ p := iface.IpfsPath(b.Cid())
81
82
out, err := apis[2].Dht().FindProviders(ctx, p, apis[2].Dht().WithNumProviders(1))
83
if err != nil {
@@ -90,8 +86,8 @@ func TestDhtProvide(t *testing.T) {
86
87
provider := <-out
88
93
- if provider.String() != "<peer.ID >" {
94
- t.Errorf("got wrong provider: %s != %s", provider.String(), nds[0].Identity.String())
89
+ if provider.ID.String() != "<peer.ID >" {
90
+ t.Errorf("got wrong provider: %s != %s", provider.ID.String(), nds[0].Identity.String())
91
}
92
93
err = apis[0].Dht().Provide(ctx, p)
@@ -106,7 +102,7 @@ func TestDhtProvide(t *testing.T) {
102
103
provider = <-out
104
109
- if provider.String() != nds[0].Identity.String() {
110
- t.Errorf("got wrong provider: %s != %s", provider.String(), nds[0].Identity.String())
105
+ if provider.ID.String() != nds[0].Identity.String() {
106
+ t.Errorf("got wrong provider: %s != %s", provider.ID.String(), nds[0].Identity.String())
107
}
108
}
core/coreapi/interface/dht.go
+4
-4
@@ -3,21 +3,21 @@ package iface
3
import (
4
"context"
5
6
- options "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
6
+ "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
7
8
peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
9
- ma "gx/ipfs/QmYmsdtJ3HsodkePE3eU3TsCaP2YvPZJ4LoXnNkDE5Tpt7/go-multiaddr"
9
+ pstore "gx/ipfs/Qmda4cPRvSRyox3SqgJN6DfSZGU5TtHufPTp9uXjFj71X6/go-libp2p-peerstore"
10
)
11
12
// DhtAPI specifies the interface to the DHT
13
type DhtAPI interface {
14
// FindPeer queries the DHT for all of the multiaddresses associated with a
15
// Peer ID
16
- FindPeer(context.Context, peer.ID) (<-chan ma.Multiaddr, error)
16
+ FindPeer(context.Context, peer.ID) (pstore.PeerInfo, error)
17
18
// FindProviders finds peers in the DHT who can provide a specific value
19
// given a key.
20
- FindProviders(context.Context, Path, ...options.DhtFindProvidersOption) (<-chan peer.ID, error) //TODO: is path the right choice here?
20
+ FindProviders(context.Context, Path, ...options.DhtFindProvidersOption) (<-chan pstore.PeerInfo, error)
21
22
// WithNumProviders is an option for FindProviders which specifies the
23
// number of peers to look for. Default is 20