Skip to content

Commit 21b1547

Browse files
StebalienStebalien
authored
Merge pull request #6292 from ipfs/experiment/provider-system-no-blocks
Introduce first strategic provider: do nothing
2 parents d4d24da + 065d783 commit 21b1547

25 files changed

+650
-654
lines changed

‎core/builder.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,11 @@ import (
44
"context"
55
"sync"
66

7-
"github.com/ipfs/go-metrics-interface"
8-
"go.uber.org/fx"
9-
107
"github.com/ipfs/go-ipfs/core/bootstrap"
118
"github.com/ipfs/go-ipfs/core/node"
9+
10+
"github.com/ipfs/go-metrics-interface"
11+
"go.uber.org/fx"
1212
)
1313

1414
type BuildCfg = node.BuildCfg // Alias for compatibility until we properly refactor the constructor interface

‎core/commands/bitswap.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,7 +233,7 @@ Trigger reprovider to announce our data to network.
233233
return ErrNotOnline
234234
}
235235

236-
err = nd.Reprovider.Trigger(req.Context)
236+
err = nd.Provider.Reprovide(req.Context)
237237
if err != nil {
238238
return err
239239
}

‎core/core.go

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ import (
2525
"github.com/ipfs/go-ipfs/pin"
2626
"github.com/ipfs/go-ipfs/provider"
2727
"github.com/ipfs/go-ipfs/repo"
28-
rp "github.com/ipfs/go-ipfs/reprovide"
2928

3029
bserv "github.com/ipfs/go-blockservice"
3130
bstore "github.com/ipfs/go-ipfs-blockstore"
@@ -92,8 +91,7 @@ type IpfsNode struct {
9291
Routing routing.IpfsRouting `optional:"true"` // the routing system. recommend ipfs-dht
9392
Exchange exchange.Interface // the block exchange + strategy (bitswap)
9493
Namesys namesys.NameSystem // the name system, resolves paths to hashes
95-
Provider provider.Provider // the value provider system
96-
Reprovider *rp.Reprovider `optional:"true"` // the value reprovider system
94+
Provider provider.System // the value provider system
9795
IpnsRepub *ipnsrp.Republisher `optional:"true"`
9896

9997
AutoNAT *autonat.AutoNATService `optional:"true"`

‎core/coreapi/coreapi.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ type CoreAPI struct {
6868
namesys namesys.NameSystem
6969
routing routing.IpfsRouting
7070

71-
provider provider.Provider
71+
provider provider.System
7272

7373
pubSub *pubsub.PubSub
7474

‎core/node/core.go

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -59,15 +59,18 @@ func Dag(bs blockservice.BlockService) format.DAGService {
5959
}
6060

6161
// OnlineExchange creates new LibP2P backed block exchange (BitSwap)
62-
func OnlineExchange(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.IpfsRouting, bs blockstore.GCBlockstore) exchange.Interface {
63-
bitswapNetwork := network.NewFromIpfsHost(host, rt)
64-
exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs)
65-
lc.Append(fx.Hook{
66-
OnStop: func(ctx context.Context) error {
67-
return exch.Close()
68-
},
69-
})
70-
return exch
62+
func OnlineExchange(provide bool) interface{} {
63+
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.IpfsRouting, bs blockstore.GCBlockstore) exchange.Interface {
64+
bitswapNetwork := network.NewFromIpfsHost(host, rt)
65+
exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, bs, bitswap.ProvideEnabled(provide))
66+
lc.Append(fx.Hook{
67+
OnStop: func(ctx context.Context) error {
68+
return exch.Close()
69+
},
70+
})
71+
return exch
72+
73+
}
7174
}
7275

7376
// Files loads persisted MFS root

‎core/node/groups.go

Lines changed: 14 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,6 @@ import (
1515

1616
"github.com/ipfs/go-ipfs/core/node/libp2p"
1717
"github.com/ipfs/go-ipfs/p2p"
18-
"github.com/ipfs/go-ipfs/provider"
19-
"github.com/ipfs/go-ipfs/reprovide"
2018

2119
offline "github.com/ipfs/go-ipfs-exchange-offline"
2220
offroute "github.com/ipfs/go-ipfs-routing/offline"
@@ -186,42 +184,6 @@ var IPNS = fx.Options(
186184
fx.Provide(RecordValidator),
187185
)
188186

189-
// Providers groups units managing provider routing records
190-
func Providers(cfg *config.Config) fx.Option {
191-
reproviderInterval := kReprovideFrequency
192-
if cfg.Reprovider.Interval != "" {
193-
dur, err := time.ParseDuration(cfg.Reprovider.Interval)
194-
if err != nil {
195-
return fx.Error(err)
196-
}
197-
198-
reproviderInterval = dur
199-
}
200-
201-
var keyProvider fx.Option
202-
switch cfg.Reprovider.Strategy {
203-
case "all":
204-
fallthrough
205-
case "":
206-
keyProvider = fx.Provide(reprovide.NewBlockstoreProvider)
207-
case "roots":
208-
keyProvider = fx.Provide(reprovide.NewPinnedProvider(true))
209-
case "pinned":
210-
keyProvider = fx.Provide(reprovide.NewPinnedProvider(false))
211-
default:
212-
return fx.Error(fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy))
213-
}
214-
215-
return fx.Options(
216-
fx.Provide(ProviderQueue),
217-
fx.Provide(ProviderCtor),
218-
fx.Provide(ReproviderCtor(reproviderInterval)),
219-
keyProvider,
220-
221-
fx.Invoke(Reprovider),
222-
)
223-
}
224-
225187
// Online groups online-only units
226188
func Online(bcfg *BuildCfg, cfg *config.Config) fx.Option {
227189

@@ -261,26 +223,31 @@ func Online(bcfg *BuildCfg, cfg *config.Config) fx.Option {
261223
recordLifetime = d
262224
}
263225

226+
/* don't provide from bitswap when the strategic provider service is active */
227+
shouldBitswapProvide := !cfg.Experimental.StrategicProviding
228+
264229
return fx.Options(
265-
fx.Provide(OnlineExchange),
230+
fx.Provide(OnlineExchange(shouldBitswapProvide)),
266231
fx.Provide(Namesys(ipnsCacheSize)),
267232

268233
fx.Invoke(IpnsRepublisher(repubPeriod, recordLifetime)),
269234

270235
fx.Provide(p2p.New),
271236

272237
LibP2P(bcfg, cfg),
273-
Providers(cfg),
238+
OnlineProviders(cfg.Experimental.StrategicProviding, cfg.Reprovider.Strategy, cfg.Reprovider.Interval),
274239
)
275240
}
276241

277242
// Offline groups offline alternatives to Online units
278-
var Offline = fx.Options(
279-
fx.Provide(offline.Exchange),
280-
fx.Provide(Namesys(0)),
281-
fx.Provide(offroute.NewOfflineRouter),
282-
fx.Provide(provider.NewOfflineProvider),
283-
)
243+
func Offline(cfg *config.Config) fx.Option {
244+
return fx.Options(
245+
fx.Provide(offline.Exchange),
246+
fx.Provide(Namesys(0)),
247+
fx.Provide(offroute.NewOfflineRouter),
248+
OfflineProviders(cfg.Experimental.StrategicProviding, cfg.Reprovider.Strategy, cfg.Reprovider.Interval),
249+
)
250+
}
284251

285252
// Core groups basic IPFS services
286253
var Core = fx.Options(
@@ -295,7 +262,7 @@ func Networked(bcfg *BuildCfg, cfg *config.Config) fx.Option {
295262
if bcfg.Online {
296263
return Online(bcfg, cfg)
297264
}
298-
return Offline
265+
return Offline(cfg)
299266
}
300267

301268
// IPFS builds a group of fx Options based on the passed BuildCfg

‎core/node/provider.go

Lines changed: 94 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -2,50 +2,117 @@ package node
22

33
import (
44
"context"
5+
"fmt"
56
"time"
67

7-
"github.com/libp2p/go-libp2p-routing"
88
"go.uber.org/fx"
99

1010
"github.com/ipfs/go-ipfs/core/node/helpers"
1111
"github.com/ipfs/go-ipfs/provider"
12+
q "github.com/ipfs/go-ipfs/provider/queue"
13+
"github.com/ipfs/go-ipfs/provider/simple"
1214
"github.com/ipfs/go-ipfs/repo"
13-
"github.com/ipfs/go-ipfs/reprovide"
15+
"github.com/libp2p/go-libp2p-routing"
1416
)
1517

1618
const kReprovideFrequency = time.Hour * 12
1719

20+
// SIMPLE
21+
1822
// ProviderQueue creates new datastore backed provider queue
19-
func ProviderQueue(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo) (*provider.Queue, error) {
20-
return provider.NewQueue(helpers.LifecycleCtx(mctx, lc), "provider-v1", repo.Datastore())
23+
func ProviderQueue(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo) (*q.Queue, error) {
24+
return q.NewQueue(helpers.LifecycleCtx(mctx, lc), "provider-v1", repo.Datastore())
25+
}
26+
27+
// SimpleProvider creates new record provider
28+
func SimpleProvider(mctx helpers.MetricsCtx, lc fx.Lifecycle, queue *q.Queue, rt routing.IpfsRouting) provider.Provider {
29+
return simple.NewProvider(helpers.LifecycleCtx(mctx, lc), queue, rt)
30+
}
31+
32+
// SimpleReprovider creates new reprovider
33+
func SimpleReprovider(reproviderInterval time.Duration) interface{} {
34+
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, rt routing.IpfsRouting, keyProvider simple.KeyChanFunc) (provider.Reprovider, error) {
35+
return simple.NewReprovider(helpers.LifecycleCtx(mctx, lc), reproviderInterval, rt, keyProvider), nil
36+
}
37+
}
38+
39+
// SimpleProviderSys creates new provider system
40+
func SimpleProviderSys(isOnline bool) interface{} {
41+
return func(lc fx.Lifecycle, p provider.Provider, r provider.Reprovider) provider.System {
42+
sys := provider.NewSystem(p, r)
43+
44+
if isOnline {
45+
lc.Append(fx.Hook{
46+
OnStart: func(ctx context.Context) error {
47+
sys.Run()
48+
return nil
49+
},
50+
OnStop: func(ctx context.Context) error {
51+
return sys.Close()
52+
},
53+
})
54+
}
55+
56+
return sys
57+
}
2158
}
2259

23-
// ProviderCtor creates new record provider
24-
func ProviderCtor(mctx helpers.MetricsCtx, lc fx.Lifecycle, queue *provider.Queue, rt routing.IpfsRouting) provider.Provider {
25-
p := provider.NewProvider(helpers.LifecycleCtx(mctx, lc), queue, rt)
26-
27-
lc.Append(fx.Hook{
28-
OnStart: func(ctx context.Context) error {
29-
p.Run()
30-
return nil
31-
},
32-
OnStop: func(ctx context.Context) error {
33-
return p.Close()
34-
},
35-
})
36-
37-
return p
60+
// ONLINE/OFFLINE
61+
62+
// OnlineProviders groups units managing provider routing records online
63+
func OnlineProviders(useStrategicProviding bool, reprovideStrategy string, reprovideInterval string) fx.Option {
64+
if useStrategicProviding {
65+
return fx.Provide(provider.NewOfflineProvider)
66+
}
67+
68+
return fx.Options(
69+
SimpleProviders(reprovideStrategy, reprovideInterval),
70+
fx.Provide(SimpleProviderSys(true)),
71+
)
3872
}
3973

40-
// ReproviderCtor creates new reprovider
41-
func ReproviderCtor(reproviderInterval time.Duration) func(helpers.MetricsCtx, fx.Lifecycle, routing.IpfsRouting, reprovide.KeyChanFunc) (*reprovide.Reprovider, error) {
42-
return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, rt routing.IpfsRouting, keyProvider reprovide.KeyChanFunc) (*reprovide.Reprovider, error) {
43-
return reprovide.NewReprovider(helpers.LifecycleCtx(mctx, lc), reproviderInterval, rt, keyProvider), nil
74+
// OfflineProviders groups units managing provider routing records offline
75+
func OfflineProviders(useStrategicProviding bool, reprovideStrategy string, reprovideInterval string) fx.Option {
76+
if useStrategicProviding {
77+
return fx.Provide(provider.NewOfflineProvider)
4478
}
79+
80+
return fx.Options(
81+
SimpleProviders(reprovideStrategy, reprovideInterval),
82+
fx.Provide(SimpleProviderSys(false)),
83+
)
4584
}
4685

47-
// Reprovider runs the reprovider service
48-
func Reprovider(lp lcProcess, reprovider *reprovide.Reprovider) error {
49-
lp.Append(reprovider.Run)
50-
return nil
86+
// SimpleProviders creates the simple provider/reprovider dependencies
87+
func SimpleProviders(reprovideStrategy string, reprovideInterval string) fx.Option {
88+
reproviderInterval := kReprovideFrequency
89+
if reprovideInterval != "" {
90+
dur, err := time.ParseDuration(reprovideInterval)
91+
if err != nil {
92+
return fx.Error(err)
93+
}
94+
95+
reproviderInterval = dur
96+
}
97+
98+
var keyProvider fx.Option
99+
switch reprovideStrategy {
100+
case "all":
101+
fallthrough
102+
case "":
103+
keyProvider = fx.Provide(simple.NewBlockstoreProvider)
104+
case "roots":
105+
keyProvider = fx.Provide(simple.NewPinnedProvider(true))
106+
case "pinned":
107+
keyProvider = fx.Provide(simple.NewPinnedProvider(false))
108+
default:
109+
return fx.Error(fmt.Errorf("unknown reprovider strategy '%s'", reprovideStrategy))
110+
}
111+
112+
return fx.Options(
113+
fx.Provide(ProviderQueue),
114+
fx.Provide(SimpleProvider),
115+
keyProvider,
116+
fx.Provide(SimpleReprovider(reproviderInterval)),
117+
)
51118
}

‎docs/experimental-features.md

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ the above issue.
2929
- [QUIC](#quic)
3030
- [AutoRelay](#autorelay)
3131
- [TLS 1.3 Handshake](#tls-13-as-default-handshake-protocol)
32+
- [Strategic Providing](#strategic-providing)
3233

3334
---
3435

@@ -674,3 +675,31 @@ ipfs config --json Experimental.PreferTLS true
674675

675676
- [ ] needs testing
676677
- [ ] needs adoption
678+
679+
---
680+
681+
## Strategic Providing
682+
683+
### State
684+
685+
Experimental, disabled by default.
686+
687+
Replaces the existing provide mechanism with a robust, strategic provider system.
688+
689+
### How to enable
690+
691+
Modify your ipfs config:
692+
693+
```
694+
ipfs config --json Experimental.StrategicProviding true
695+
```
696+
697+
### Road to being a real feature
698+
699+
- [ ] needs real world testing
700+
- [ ] needs adoption
701+
- [ ] needs to support all providing features
702+
- [X] provide nothing
703+
- [ ] provide roots
704+
- [ ] provide all
705+
- [ ] provide strategic

‎go.mod

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ require (
3535
github.com/ipfs/go-ipfs-blocksutil v0.0.1
3636
github.com/ipfs/go-ipfs-chunker v0.0.1
3737
github.com/ipfs/go-ipfs-cmds v0.0.8
38-
github.com/ipfs/go-ipfs-config v0.0.3
38+
github.com/ipfs/go-ipfs-config v0.0.4
3939
github.com/ipfs/go-ipfs-ds-help v0.0.1
4040
github.com/ipfs/go-ipfs-exchange-interface v0.0.1
4141
github.com/ipfs/go-ipfs-exchange-offline v0.0.1
@@ -122,6 +122,7 @@ require (
122122
go.uber.org/multierr v1.1.0 // indirect
123123
go4.org v0.0.0-20190313082347-94abd6928b1d // indirect
124124
golang.org/x/sys v0.0.0-20190522044717-8097e1b27ff5
125+
google.golang.org/appengine v1.4.0 // indirect
125126
gopkg.in/cheggaaa/pb.v1 v1.0.28
126127
gotest.tools/gotestsum v0.3.4
127128
)

0 commit comments

Comments
 (0)