move streaming set to thirdparty
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Aug 10, 2018 at 13:24 UTC
e1bdf6cd280bcbe90b08b71500aa86e0a29a2217
4 files changed
+56
-45
core/coreapi/dht.go
+7
-13
@@ -100,26 +100,20 @@ func provideKeys(ctx context.Context, r routing.IpfsRouting, cids []*cid.Cid) er
100
func provideKeysRec(ctx context.Context, r routing.IpfsRouting, bs blockstore.Blockstore, cids []*cid.Cid) error {
101
provided := cid.NewSet()
102
for _, c := range cids {
103
- kset := cid.NewSet()
104
-
103
dserv := dag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
104
107
- err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, kset.Visit)
105
+ err := dag.EnumerateChildrenAsync(ctx, dag.GetLinksDirect(dserv), c, provided.Visit)
106
if err != nil {
107
return err
108
}
109
+ }
110
112
- for _, k := range kset.Keys() {
113
- if provided.Has(k) {
114
- continue
115
- }
116
-
117
- err = r.Provide(ctx, k, true)
118
- if err != nil {
119
- return err
120
- }
121
- provided.Add(k)
111
+ for _, k := range provided.Keys() {
112
+ err := r.Provide(ctx, k, true)
113
+ if err != nil {
114
+ return err
115
}
116
+ provided.Add(k)
117
}
118
119
return nil
core/coreapi/interface/dht.go
+1
-1
@@ -10,7 +10,7 @@ import (
10
)
11
12
// DhtAPI specifies the interface to the DHT
13
-// Note: This API will likely get renamed in near future, see
13
+// Note: This API will likely get deprecated in near future, see
14
// https://github.com/ipfs/interface-ipfs-core/issues/249 for more context.
15
type DhtAPI interface {
16
// FindPeer queries the DHT for all of the multiaddresses associated with a
exchange/reprovide/providers.go
+10
-31
@@ -4,6 +4,7 @@ import (
4
"context"
5
6
pin "github.com/ipfs/go-ipfs/pin"
7
+ "github.com/ipfs/go-ipfs/thirdparty/streaming-cid-set"
8
9
merkledag "gx/ipfs/QmNr4E8z9bGTztvHJktp7uQaMdx9p3r9Asrq6eYk7iCh4a/go-merkledag"
10
ipld "gx/ipfs/QmX5CsuHyVZeTLxgRSYkgLSDQKb9UjE8xnhQzCEJWWWFsC/go-ipld-format"
@@ -29,7 +30,7 @@ func NewPinnedProvider(pinning pin.Pinner, dag ipld.DAGService, onlyRoots bool)
30
outCh := make(chan *cid.Cid)
31
go func() {
32
defer close(outCh)
32
- for c := range set.new {
33
+ for c := range set.New {
34
select {
35
case <-ctx.Done():
36
return
@@ -43,21 +44,23 @@ func NewPinnedProvider(pinning pin.Pinner, dag ipld.DAGService, onlyRoots bool)
44
}
45
}
46
46
-func pinSet(ctx context.Context, pinning pin.Pinner, dag ipld.DAGService, onlyRoots bool) (*streamingSet, error) {
47
- set := newStreamingSet()
47
+func pinSet(ctx context.Context, pinning pin.Pinner, dag ipld.DAGService, onlyRoots bool) (*streamingset.StreamingSet, error) {
48
+ set := streamingset.NewStreamingSet()
49
50
go func() {
50
- defer close(set.new)
51
+ ctx, cancel := context.WithCancel(ctx)
52
+ defer cancel()
53
+ defer close(set.New)
54
55
for _, key := range pinning.DirectKeys() {
53
- set.add(key)
56
+ set.Visitor(ctx)(key)
57
}
58
59
for _, key := range pinning.RecursiveKeys() {
57
- set.add(key)
60
+ set.Visitor(ctx)(key)
61
62
if !onlyRoots {
60
- err := merkledag.EnumerateChildren(ctx, merkledag.GetLinksWithDAG(dag), key, set.add)
63
+ err := merkledag.EnumerateChildren(ctx, merkledag.GetLinksWithDAG(dag), key, set.Visitor(ctx))
64
if err != nil {
65
log.Errorf("reprovide indirect pins: %s", err)
66
return
@@ -68,27 +71,3 @@ func pinSet(ctx context.Context, pinning pin.Pinner, dag ipld.DAGService, onlyRo
71
72
return set, nil
73
}
71
-
72
-type streamingSet struct {
73
- set *cid.Set
74
- new chan *cid.Cid
75
-}
76
-
77
-// NewSet initializes and returns a new Set.
78
-func newStreamingSet() *streamingSet {
79
- return &streamingSet{
80
- set: cid.NewSet(),
81
- new: make(chan *cid.Cid),
82
- }
83
-}
84
-
85
-// add adds a Cid to the set only if it is
86
-// not in it already.
87
-func (s *streamingSet) add(c *cid.Cid) bool {
88
- if s.set.Visit(c) {
89
- s.new <- c
90
- return true
91
- }
92
-
93
- return false
94
-}
thirdparty/streaming-cid-set/set.go
new
+38
@@ -0,0 +1,38 @@
1
+package streamingset
2
+
3
+import (
4
+ "context"
5
+
6
+ cid "gx/ipfs/QmZFbDTY9jfSBms2MchvYM9oYRbAF19K7Pby47yDBfpPrb/go-cid"
7
+)
8
+
9
+// StreamingSet is an extension of cid.Set which allows to implement back-pressure
10
+// for the Visit function
11
+type StreamingSet struct {
12
+ Set *cid.Set
13
+ New chan *cid.Cid
14
+}
15
+
16
+// NewStreamingSet initializes and returns new Set.
17
+func NewStreamingSet() *StreamingSet {
18
+ return &StreamingSet{
19
+ Set: cid.NewSet(),
20
+ New: make(chan *cid.Cid),
21
+ }
22
+}
23
+
24
+// Visitor creates new visitor which adds a Cids to the set and emits them to
25
+// the set.New channel
26
+func (s *StreamingSet) Visitor(ctx context.Context) func(c *cid.Cid) bool {
27
+ return func(c *cid.Cid) bool {
28
+ if s.Set.Visit(c) {
29
+ select {
30
+ case s.New <- c:
31
+ case <-ctx.Done():
32
+ }
33
+ return true
34
+ }
35
+
36
+ return false
37
+ }
38
+}