reprovider: apply review suggestions
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Aug 15, 2017 at 23:03 UTC
4a5b93a1df25dd6694f514a60a0c1e94ba10a74d
4 files changed
+18
-21
core/core.go
+2
-2
@@ -263,7 +263,7 @@ func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
263
return err
264
}
265
266
- var keyProvider func(context.Context) (<-chan *cid.Cid, error)
266
+ var keyProvider rp.KeyChanFunc
267
268
switch cfg.Reprovider.Strategy {
269
case "all":
@@ -275,7 +275,7 @@ func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
275
case "pinned":
276
keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, false)
277
default:
278
- return fmt.Errorf("unknown reprovider strtaegy '%s'", cfg.Reprovider.Strategy)
278
+ return fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
279
}
280
n.Reprovider = rp.NewReprovider(ctx, n.Routing, keyProvider)
281
docs/config.md
+1
-1
@@ -15,7 +15,7 @@ a running daemon do not read the config file at runtime.
15
- [`Identity`](#identity)
16
- [`Ipns`](#ipns)
17
- [`Mounts`](#mounts)
18
-- [`Reproviderl`](#reprovider)
18
+- [`Reprovider`](#reprovider)
19
- [`SupernodeRouting`](#supernoderouting)
20
- [`Swarm`](#swarm)
21
- [`Tour`](#tour)
exchange/reprovide/providers.go
+9
-15
@@ -11,14 +11,14 @@ import (
11
)
12
13
// NewBlockstoreProvider returns key provider using bstore.AllKeysChan
14
-func NewBlockstoreProvider(bstore blocks.Blockstore) keyChanFunc {
14
+func NewBlockstoreProvider(bstore blocks.Blockstore) KeyChanFunc {
15
return func(ctx context.Context) (<-chan *cid.Cid, error) {
16
return bstore.AllKeysChan(ctx)
17
}
18
}
19
20
// NewPinnedProvider returns provider supplying pinned keys
21
-func NewPinnedProvider(pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) keyChanFunc {
21
+func NewPinnedProvider(pinning pin.Pinner, dag merkledag.DAGService, onlyRoots bool) KeyChanFunc {
22
return func(ctx context.Context) (<-chan *cid.Cid, error) {
23
set, err := pinSet(ctx, pinning, dag, onlyRoots)
24
if err != nil {
@@ -46,6 +46,8 @@ func pinSet(ctx context.Context, pinning pin.Pinner, dag merkledag.DAGService, o
46
set := newStreamingSet()
47
48
go func() {
49
+ defer close(set.new)
50
+
51
for _, key := range pinning.DirectKeys() {
52
set.add(key)
53
}
@@ -56,41 +58,33 @@ func pinSet(ctx context.Context, pinning pin.Pinner, dag merkledag.DAGService, o
58
if !onlyRoots {
59
err := merkledag.EnumerateChildren(ctx, dag.GetLinks, key, set.add)
60
if err != nil {
59
- return //TODO: propagate to chan / log?
61
+ log.Errorf("reprovide indirect pins: %s", err)
62
+ return
63
}
64
}
65
}
63
-
64
- close(set.new)
66
}()
67
68
return set, nil
69
}
70
71
type streamingSet struct {
71
- set map[string]struct{}
72
+ set *cid.Set
73
new chan *cid.Cid
74
}
75
76
// NewSet initializes and returns a new Set.
77
func newStreamingSet() *streamingSet {
78
return &streamingSet{
78
- set: make(map[string]struct{}),
79
+ set: cid.NewSet(),
80
new: make(chan *cid.Cid),
81
}
82
}
83
83
-// has returns if the Set contains a given Cid.
84
-func (s *streamingSet) has(c *cid.Cid) bool {
85
- _, ok := s.set[string(c.Bytes())]
86
- return ok
87
-}
88
-
84
// add adds a Cid to the set only if it is
85
// not in it already.
86
func (s *streamingSet) add(c *cid.Cid) bool {
92
- if !s.has(c) {
93
- s.set[string(c.Bytes())] = struct{}{}
87
+ if s.set.Visit(c) {
88
s.new <- c
89
return true
90
}
exchange/reprovide/reprovide.go
+6
-3
@@ -13,7 +13,8 @@ import (
13
14
var log = logging.Logger("reprovider")
15
16
-type keyChanFunc func(context.Context) (<-chan *cid.Cid, error)
16
+//KeyChanFunc is function streaming CIDs to pass to content routing
17
+type KeyChanFunc func(context.Context) (<-chan *cid.Cid, error)
18
type doneFunc func(error)
19
20
type Reprovider struct {
@@ -23,11 +24,11 @@ type Reprovider struct {
24
// The routing system to provide values through
25
rsys routing.ContentRouting
26
26
- keyProvider keyChanFunc
27
+ keyProvider KeyChanFunc
28
}
29
30
// NewReprovider creates new Reprovider instance.
30
-func NewReprovider(ctx context.Context, rsys routing.ContentRouting, keyProvider keyChanFunc) *Reprovider {
31
+func NewReprovider(ctx context.Context, rsys routing.ContentRouting, keyProvider KeyChanFunc) *Reprovider {
32
return &Reprovider{
33
ctx: ctx,
34
trigger: make(chan doneFunc),
@@ -52,6 +53,8 @@ func (rp *Reprovider) ProvideEvery(tick time.Duration) {
53
case <-after:
54
}
55
56
+ //'mute' the trigger channel so when `ipfs bitswap reprovide` is called
57
+ //a 'reprovider is already running' error is returned
58
unmute := rp.muteTrigger()
59
60
err := rp.Reprovide()