Fix context propagation sortof
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Apr 1, 2019 at 18:39 UTC
0ba7661d2056a00cef5ab76f7395b4d4a318a2eb
2 files changed
+35
-28
core/builder.go
+7
@@ -129,6 +129,8 @@ func defaultRepo(dstore repo.Datastore) (repo.Repo, error) {
129
}, nil
130
}
131
132
+type MetricsCtx context.Context
133
+
134
// NewNode constructs and returns an IpfsNode using the given cfg.
135
func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
136
if cfg == nil {
@@ -157,9 +159,14 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
159
return cfg
160
})
161
162
+ metricsCtx := fx.Provide(func() MetricsCtx {
163
+ return MetricsCtx(ctx)
164
+ })
165
+
166
params := fx.Options(
167
repoOption,
168
cfgOption,
169
+ metricsCtx,
170
)
171
172
storage := fx.Options(
core/ncore.go
+28
-28
@@ -117,7 +117,7 @@ func datastoreCtor(repo repo.Repo) ds.Datastore {
117
118
type BaseBlocks bstore.Blockstore
119
120
-func baseBlockstoreCtor(repo repo.Repo, cfg *iconfig.Config, bcfg *BuildCfg, lc fx.Lifecycle) (bs BaseBlocks, err error) {
120
+func baseBlockstoreCtor(mctx MetricsCtx, repo repo.Repo, cfg *iconfig.Config, bcfg *BuildCfg, lc fx.Lifecycle) (bs BaseBlocks, err error) {
121
rds := &retry.Datastore{
122
Batching: repo.Datastore(),
123
Delay: time.Millisecond * 200,
@@ -135,7 +135,7 @@ func baseBlockstoreCtor(repo repo.Repo, cfg *iconfig.Config, bcfg *BuildCfg, lc
135
}
136
137
if !bcfg.NilRepo {
138
- ctx, cancel := context.WithCancel(context.TODO()) //TODO: needed for mertics
138
+ ctx, cancel := context.WithCancel(mctx)
139
140
lc.Append(fx.Hook{
141
OnStop: func(context context.Context) error {
@@ -395,13 +395,13 @@ type p2pHostOut struct {
395
}
396
397
// TODO: move some of this into params struct
398
-func p2pHost(lc fx.Lifecycle, params p2pHostIn) (out p2pHostOut, err error) {
398
+func p2pHost(mctx MetricsCtx, lc fx.Lifecycle, params p2pHostIn) (out p2pHostOut, err error) {
399
opts := []libp2p.Option{libp2p.NoListenAddrs}
400
for _, o := range params.Opts {
401
opts = append(opts, o...)
402
}
403
404
- ctx, cancel := context.WithCancel(context.TODO())
404
+ ctx, cancel := context.WithCancel(mctx)
405
lc.Append(fx.Hook{
406
OnStop: func(_ context.Context) error {
407
cancel()
@@ -471,12 +471,12 @@ type p2pRoutingOut struct {
471
PSRouter *psrouter.PubsubValueStore //TODO: optional
472
}
473
474
-func p2pOnlineRouting(lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
474
+func p2pOnlineRouting(mctx MetricsCtx, lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
475
out.IpfsRouting = in.BaseRouting
476
477
if in.BCfg.getOpt("ipnsps") {
478
out.PSRouter = psrouter.NewPubsubValueStore(
479
- lifecycleCtx(lc),
479
+ lifecycleCtx(mctx, lc),
480
in.Host,
481
in.BaseRouting,
482
in.PubSub,
@@ -503,7 +503,7 @@ func p2pOnlineRouting(lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
503
////////////
504
// P2P services
505
506
-func autoNATService(lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host) error {
506
+func autoNATService(mctx MetricsCtx, lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host) error {
507
if !cfg.Swarm.EnableAutoNATService {
508
return nil
509
}
@@ -512,11 +512,11 @@ func autoNATService(lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host) err
512
opts = append(opts, libp2p.DefaultTransports, libp2p.Transport(quic.NewTransport))
513
}
514
515
- _, err := autonat.NewAutoNATService(lifecycleCtx(lc), host, opts...)
515
+ _, err := autonat.NewAutoNATService(lifecycleCtx(mctx, lc), host, opts...)
516
return err
517
}
518
519
-func pubsubCtor(lc fx.Lifecycle, host p2phost.Host, bcfg *BuildCfg, cfg *iconfig.Config) (service *pubsub.PubSub, err error) {
519
+func pubsubCtor(mctx MetricsCtx, lc fx.Lifecycle, host p2phost.Host, bcfg *BuildCfg, cfg *iconfig.Config) (service *pubsub.PubSub, err error) {
520
if !(bcfg.getOpt("pubsub") || bcfg.getOpt("ipnsps")) {
521
return nil, nil // TODO: mark optional
522
}
@@ -534,10 +534,10 @@ func pubsubCtor(lc fx.Lifecycle, host p2phost.Host, bcfg *BuildCfg, cfg *iconfig
534
case "":
535
fallthrough
536
case "floodsub":
537
- service, err = pubsub.NewFloodSub(lifecycleCtx(lc), host, pubsubOptions...)
537
+ service, err = pubsub.NewFloodSub(lifecycleCtx(mctx, lc), host, pubsubOptions...)
538
539
case "gossipsub":
540
- service, err = pubsub.NewGossipSub(lifecycleCtx(lc), host, pubsubOptions...)
540
+ service, err = pubsub.NewGossipSub(lifecycleCtx(mctx, lc), host, pubsubOptions...)
541
542
default:
543
err = fmt.Errorf("Unknown pubsub router %s", cfg.Pubsub.Router)
@@ -577,9 +577,9 @@ func dagCtor(bs bserv.BlockService) format.DAGService {
577
return merkledag.NewDAGService(bs)
578
}
579
580
-func onlineExchangeCtor(lc fx.Lifecycle, host p2phost.Host, rt routing.IpfsRouting, bs bstore.GCBlockstore) exchange.Interface {
580
+func onlineExchangeCtor(mctx MetricsCtx, lc fx.Lifecycle, host p2phost.Host, rt routing.IpfsRouting, bs bstore.GCBlockstore) exchange.Interface {
581
bitswapNetwork := bsnet.NewFromIpfsHost(host, rt)
582
- return bitswap.New(lifecycleCtx(lc), bitswapNetwork, bs)
582
+ return bitswap.New(lifecycleCtx(mctx, lc), bitswapNetwork, bs)
583
}
584
585
func onlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo, cfg *iconfig.Config) (namesys.NameSystem, error) {
@@ -636,20 +636,20 @@ func (dh *discoveryHandler) HandlePeerFound(p pstore.PeerInfo) {
636
}
637
}
638
639
-func newDiscoveryHandler(lc fx.Lifecycle, host p2phost.Host) *discoveryHandler {
639
+func newDiscoveryHandler(mctx MetricsCtx, lc fx.Lifecycle, host p2phost.Host) *discoveryHandler {
640
return &discoveryHandler{
641
- ctx: lifecycleCtx(lc),
641
+ ctx: lifecycleCtx(mctx, lc),
642
host: host,
643
}
644
}
645
646
-func setupDiscovery(lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host, handler *discoveryHandler) error {
646
+func setupDiscovery(mctx MetricsCtx, lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host, handler *discoveryHandler) error {
647
if cfg.Discovery.MDNS.Enabled {
648
mdns := cfg.Discovery.MDNS
649
if mdns.Interval == 0 {
650
mdns.Interval = 5
651
}
652
- service, err := discovery.NewMdnsService(lifecycleCtx(lc), host, time.Duration(mdns.Interval)*time.Second, discovery.ServiceTag)
652
+ service, err := discovery.NewMdnsService(lifecycleCtx(mctx, lc), host, time.Duration(mdns.Interval)*time.Second, discovery.ServiceTag)
653
if err != nil {
654
log.Error("mdns error: ", err)
655
return nil
@@ -659,15 +659,15 @@ func setupDiscovery(lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host, han
659
return nil
660
}
661
662
-func providerQueue(lc fx.Lifecycle, repo repo.Repo) (*provider.Queue, error) {
663
- return provider.NewQueue(lifecycleCtx(lc), "provider-v1", repo.Datastore())
662
+func providerQueue(mctx MetricsCtx, lc fx.Lifecycle, repo repo.Repo) (*provider.Queue, error) {
663
+ return provider.NewQueue(lifecycleCtx(mctx, lc), "provider-v1", repo.Datastore())
664
}
665
666
-func providerCtor(lc fx.Lifecycle, queue *provider.Queue, rt routing.IpfsRouting) provider.Provider {
667
- return provider.NewProvider(lifecycleCtx(lc), queue, rt)
666
+func providerCtor(mctx MetricsCtx, lc fx.Lifecycle, queue *provider.Queue, rt routing.IpfsRouting) provider.Provider {
667
+ return provider.NewProvider(lifecycleCtx(mctx, lc), queue, rt)
668
}
669
670
-func reproviderCtor(lc fx.Lifecycle, cfg *iconfig.Config, bs BaseBlocks, ds format.DAGService, pinning pin.Pinner, rt routing.IpfsRouting) (*rp.Reprovider, error) {
670
+func reproviderCtor(mctx MetricsCtx, lc fx.Lifecycle, cfg *iconfig.Config, bs BaseBlocks, ds format.DAGService, pinning pin.Pinner, rt routing.IpfsRouting) (*rp.Reprovider, error) {
671
var keyProvider rp.KeyChanFunc
672
673
switch cfg.Reprovider.Strategy {
@@ -682,7 +682,7 @@ func reproviderCtor(lc fx.Lifecycle, cfg *iconfig.Config, bs BaseBlocks, ds form
682
default:
683
return nil, fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
684
}
685
- return rp.NewReprovider(lifecycleCtx(lc), rt, keyProvider), nil
685
+ return rp.NewReprovider(lifecycleCtx(mctx, lc), rt, keyProvider), nil
686
}
687
688
func reprovider(cfg *iconfig.Config, reprovider *rp.Reprovider) error {
@@ -700,7 +700,7 @@ func reprovider(cfg *iconfig.Config, reprovider *rp.Reprovider) error {
700
return nil
701
}
702
703
-func files(lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
703
+func files(mctx MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
704
dsk := ds.NewKey("/local/filesroot")
705
pf := func(ctx context.Context, c cid.Cid) error {
706
return repo.Datastore().Put(dsk, c.Bytes())
@@ -708,7 +708,7 @@ func files(lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, e
708
709
var nd *merkledag.ProtoNode
710
val, err := repo.Datastore().Get(dsk)
711
- ctx := lifecycleCtx(lc)
711
+ ctx := lifecycleCtx(mctx, lc)
712
713
switch {
714
case err == ds.ErrNotFound || val == nil:
@@ -748,8 +748,8 @@ func files(lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, e
748
//
749
// This is a hack which we need because most of our services use contexts in a
750
// wrong way
751
-func lifecycleCtx(lc fx.Lifecycle) context.Context {
752
- ctx, cancel := context.WithCancel(context.TODO()) // TODO: really wire this context up, things (like metrics) may depend on it
751
+func lifecycleCtx(mctx MetricsCtx, lc fx.Lifecycle) context.Context {
752
+ ctx, cancel := context.WithCancel(mctx)
753
lc.Append(fx.Hook{
754
OnStop: func(_ context.Context) error {
755
cancel()
@@ -774,7 +774,7 @@ func (lp *lcProcess) Run(f goprocess.ProcessFunc) {
774
return nil
775
},
776
OnStop: func(ctx context.Context) error {
777
- return (<-proc).Close() // todo: respect ctx
777
+ return (<-proc).Close() // todo: respect ctx, somehow
778
},
779
})
780
}