reprovider: use goprocess
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Apr 23, 2019 at 20:11 UTC
d36b6dbd06b105b61ac8e5bf40c1eae6fcd474d8
4 files changed
+50
-39
core/node/provider.go
+13
-13
@@ -45,6 +45,16 @@ func ProviderCtor(mctx helpers.MetricsCtx, lc fx.Lifecycle, queue *provider.Queu
45
func ReproviderCtor(mctx helpers.MetricsCtx, lc fx.Lifecycle, cfg *config.Config, bs BaseBlocks, ds format.DAGService, pinning pin.Pinner, rt routing.IpfsRouting) (*reprovide.Reprovider, error) {
46
var keyProvider reprovide.KeyChanFunc
47
48
+ reproviderInterval := kReprovideFrequency
49
+ if cfg.Reprovider.Interval != "" {
50
+ dur, err := time.ParseDuration(cfg.Reprovider.Interval)
51
+ if err != nil {
52
+ return nil, err
53
+ }
54
+
55
+ reproviderInterval = dur
56
+ }
57
+
58
switch cfg.Reprovider.Strategy {
59
case "all":
60
fallthrough
@@ -57,21 +67,11 @@ func ReproviderCtor(mctx helpers.MetricsCtx, lc fx.Lifecycle, cfg *config.Config
67
default:
68
return nil, fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
69
}
60
- return reprovide.NewReprovider(helpers.LifecycleCtx(mctx, lc), rt, keyProvider), nil
70
+ return reprovide.NewReprovider(helpers.LifecycleCtx(mctx, lc), reproviderInterval, rt, keyProvider), nil
71
}
72
73
// Reprovider runs the reprovider service
64
-func Reprovider(cfg *config.Config, reprovider *reprovide.Reprovider) error {
65
- reproviderInterval := kReprovideFrequency
66
- if cfg.Reprovider.Interval != "" {
67
- dur, err := time.ParseDuration(cfg.Reprovider.Interval)
68
- if err != nil {
69
- return err
70
- }
71
-
72
- reproviderInterval = dur
73
- }
74
-
75
- go reprovider.Run(reproviderInterval) // TODO: refactor reprovider to have Start/Stop, use lifecycle
74
+func Reprovider(lp lcProcess, reprovider *reprovide.Reprovider) error {
75
+ lp.Append(reprovider.Run)
76
return nil
77
}
reprovide/providers.go
+4
-4
@@ -3,13 +3,13 @@ package reprovide
3
import (
4
"context"
5
6
- "github.com/ipfs/go-ipfs/pin"
6
+ pin "github.com/ipfs/go-ipfs/pin"
7
8
- "github.com/ipfs/go-cid"
9
- "github.com/ipfs/go-cidutil"
8
+ cid "github.com/ipfs/go-cid"
9
+ cidutil "github.com/ipfs/go-cidutil"
10
blocks "github.com/ipfs/go-ipfs-blockstore"
11
ipld "github.com/ipfs/go-ipld-format"
12
- "github.com/ipfs/go-merkledag"
12
+ merkledag "github.com/ipfs/go-merkledag"
13
)
14
15
// NewBlockstoreProvider returns key provider using bstore.AllKeysChan
reprovide/reprovide.go
+29
-17
@@ -2,67 +2,77 @@ package reprovide
2
3
import (
4
"context"
5
+ "errors"
6
"fmt"
7
"time"
8
8
- backoff "github.com/cenkalti/backoff"
9
- cid "github.com/ipfs/go-cid"
9
+ "github.com/cenkalti/backoff"
10
+ "github.com/ipfs/go-cid"
11
logging "github.com/ipfs/go-log"
12
"github.com/ipfs/go-verifcid"
13
+ "github.com/jbenet/goprocess"
14
+ goprocessctx "github.com/jbenet/goprocess/context"
15
routing "github.com/libp2p/go-libp2p-routing"
16
)
17
18
var log = logging.Logger("reprovider")
19
17
-//KeyChanFunc is function streaming CIDs to pass to content routing
20
+// KeyChanFunc is function streaming CIDs to pass to content routing
21
type KeyChanFunc func(context.Context) (<-chan cid.Cid, error)
22
type doneFunc func(error)
23
24
type Reprovider struct {
25
ctx context.Context
26
trigger chan doneFunc
27
+ closing chan struct{}
28
29
// The routing system to provide values through
30
rsys routing.ContentRouting
31
32
keyProvider KeyChanFunc
33
+ tick time.Duration
34
}
35
36
// NewReprovider creates new Reprovider instance.
32
-func NewReprovider(ctx context.Context, rsys routing.ContentRouting, keyProvider KeyChanFunc) *Reprovider {
37
+func NewReprovider(ctx context.Context, tick time.Duration, rsys routing.ContentRouting, keyProvider KeyChanFunc) *Reprovider {
38
return &Reprovider{
39
ctx: ctx,
40
trigger: make(chan doneFunc),
41
+ closing: make(chan struct{}),
42
43
rsys: rsys,
44
keyProvider: keyProvider,
45
+ tick: tick,
46
}
47
}
48
49
// Run re-provides keys with 'tick' interval or when triggered
43
-func (rp *Reprovider) Run(tick time.Duration) {
50
+func (rp *Reprovider) Run(proc goprocess.Process) {
51
+ ctx := goprocessctx.WithProcessClosing(rp.ctx, proc)
52
+ defer close(rp.closing)
53
+
54
// dont reprovide immediately.
55
// may have just started the daemon and shutting it down immediately.
56
// probability( up another minute | uptime ) increases with uptime.
57
after := time.After(time.Minute)
58
var done doneFunc
59
for {
50
- if tick == 0 {
60
+ if rp.tick == 0 {
61
after = make(chan time.Time)
62
}
63
64
select {
55
- case <-rp.ctx.Done():
65
+ case <-ctx.Done():
66
return
67
case done = <-rp.trigger:
68
case <-after:
69
}
70
61
- //'mute' the trigger channel so when `ipfs bitswap reprovide` is called
62
- //a 'reprovider is already running' error is returned
71
+ // 'mute' the trigger channel so when `ipfs bitswap reprovide` is called
72
+ // a 'reprovider is already running' error is returned
73
unmute := rp.muteTrigger()
74
65
- err := rp.Reprovide()
75
+ err := rp.reprovide(ctx)
76
if err != nil {
77
log.Debug(err)
78
}
@@ -73,13 +83,13 @@ func (rp *Reprovider) Run(tick time.Duration) {
83
84
unmute()
85
76
- after = time.After(tick)
86
+ after = time.After(rp.tick)
87
}
88
}
89
80
-// Reprovide registers all keys given by rp.keyProvider to libp2p content routing
81
-func (rp *Reprovider) Reprovide() error {
82
- keychan, err := rp.keyProvider(rp.ctx)
90
+// reprovide registers all keys given by rp.keyProvider to libp2p content routing
91
+func (rp *Reprovider) reprovide(ctx context.Context) error {
92
+ keychan, err := rp.keyProvider(ctx)
93
if err != nil {
94
return fmt.Errorf("failed to get key chan: %s", err)
95
}
@@ -90,7 +100,7 @@ func (rp *Reprovider) Reprovide() error {
100
continue
101
}
102
op := func() error {
93
- err := rp.rsys.Provide(rp.ctx, c, true)
103
+ err := rp.rsys.Provide(ctx, c, true)
104
if err != nil {
105
log.Debugf("Failed to provide key: %s", err)
106
}
@@ -119,10 +129,12 @@ func (rp *Reprovider) Trigger(ctx context.Context) error {
129
}
130
131
select {
132
+ case <-rp.closing:
133
+ return errors.New("reprovider is closed")
134
case <-rp.ctx.Done():
123
- return context.Canceled
135
+ return rp.ctx.Err()
136
case <-ctx.Done():
125
- return context.Canceled
137
+ return ctx.Err()
138
case rp.trigger <- df:
139
<-progressCtx.Done()
140
return err
reprovide/reprovide_test.go
+4
-5
@@ -1,8 +1,7 @@
1
-package reprovide_test
1
+package reprovide
2
3
import (
4
"context"
5
- "github.com/ipfs/go-ipfs"
5
"testing"
6
7
blocks "github.com/ipfs/go-block-format"
@@ -34,9 +33,9 @@ func TestReprovide(t *testing.T) {
33
t.Fatal(err)
34
}
35
37
- keyProvider := ipfs.NewBlockstoreProvider(bstore)
38
- reprov := ipfs.NewReprovider(ctx, clA, keyProvider)
39
- err = reprov.Reprovide()
36
+ keyProvider := NewBlockstoreProvider(bstore)
37
+ reprov := NewReprovider(ctx, 0, clA, keyProvider)
38
+ err = reprov.reprovide(ctx)
39
if err != nil {
40
t.Fatal(err)
41
}