Fix goprocess / lifecycle / ctx relations
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Mar 29, 2019 at 15:55 UTC
cc2be2e73a5f568643680f876ba3ff377d67f08b
8 files changed
+47
-25
cmd/ipfs/daemon.go
+1
-1
@@ -372,7 +372,7 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
372
if err != nil {
373
return err
374
}
375
- node.Process().AddChild(goprocess.WithTeardown(cctx.Plugins.Close))
375
+ node.Process.AddChild(goprocess.WithTeardown(cctx.Plugins.Close))
376
377
// construct api endpoint - every time
378
apiErrc, err := serveHTTPApi(req, cctx)
core/builder.go
+13
-3
@@ -5,13 +5,15 @@ import (
5
"crypto/rand"
6
"encoding/base64"
7
"errors"
8
- "github.com/ipfs/go-ipfs/p2p"
9
- "github.com/ipfs/go-ipfs/provider"
10
- "go.uber.org/fx"
8
"os"
9
"syscall"
10
"time"
11
12
+ "go.uber.org/fx"
13
+
14
+ "github.com/ipfs/go-ipfs/p2p"
15
+ "github.com/ipfs/go-ipfs/provider"
16
+
17
filestore "github.com/ipfs/go-ipfs/filestore"
18
namesys "github.com/ipfs/go-ipfs/namesys"
19
pin "github.com/ipfs/go-ipfs/pin"
@@ -219,6 +221,8 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
221
}
222
223
app := fx.New(
224
+ fx.Provide(baseProcess),
225
+
226
params,
227
storage,
228
ident,
@@ -226,12 +230,18 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
230
online,
231
232
fx.Invoke(setupSharding),
233
+ fx.NopLogger,
234
235
core,
236
237
fx.Extract(n),
238
)
239
240
+ go func() {
241
+ <-ctx.Done()
242
+ app.Stop(context.Background())
243
+ }()
244
+
245
n.IsOnline = cfg.Online
246
n.app = app
247
core/commands/shutdown.go
+1
-1
@@ -21,7 +21,7 @@ var daemonShutdownCmd = &cmds.Command{
21
return cmdkit.Errorf(cmdkit.ErrClient, "daemon not running")
22
}
23
24
- if err := nd.Process().Close(); err != nil {
24
+ if err := nd.Close(); err != nil {
25
log.Error("error while shutting down ipfs daemon:", err)
26
}
27
core/core.go
+5
-10
@@ -135,7 +135,7 @@ type IpfsNode struct {
135
DHT *dht.IpfsDHT `optional:"true"`
136
P2P *p2p.P2P `optional:"true"`
137
138
- proc goprocess.Process //TODO: remove
138
+ Process goprocess.Process
139
ctx context.Context
140
141
app *fx.App
@@ -206,9 +206,9 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
206
log.Warning("This might be configuration mistake.")
207
}
208
}
209
- case <-n.Process().Closing():
210
- t.Stop()
211
- return
209
+ //case <-n.Process().Closing():
210
+ // t.Stop()
211
+ // return
212
}
213
}
214
}()
@@ -642,16 +642,11 @@ func (n *IpfsNode) setupIpnsRepublisher() error {
642
n.IpnsRepub.RecordLifetime = d
643
}
644
645
- n.Process().Go(n.IpnsRepub.Run)
645
+ //n.Process().Go(n.IpnsRepub.Run)
646
647
return nil
648
}
649
650
-// Process returns the Process object
651
-func (n *IpfsNode) Process() goprocess.Process {
652
- return n.proc
653
-}
654
-
650
// Close calls Close() on the App object
651
func (n *IpfsNode) Close() error {
652
return n.app.Stop(n.ctx)
core/corehttp/corehttp.go
+3
-3
@@ -85,7 +85,7 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
85
}
86
87
select {
88
- case <-node.Process().Closing():
88
+ case <-node.Process.Closing():
89
return fmt.Errorf("failed to start server, process closing")
90
default:
91
}
@@ -95,7 +95,7 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
95
}
96
97
var serverError error
98
- serverProc := node.Process().Go(func(p goprocess.Process) {
98
+ serverProc := node.Process.Go(func(p goprocess.Process) {
99
serverError = server.Serve(lis)
100
})
101
@@ -103,7 +103,7 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
103
select {
104
case <-serverProc.Closed():
105
// if node being closed before server exits, close server
106
- case <-node.Process().Closing():
106
+ case <-node.Process.Closing():
107
log.Infof("server at %s terminating...", addr)
108
109
warnProc := periodicproc.Tick(5*time.Second, func(_ goprocess.Process) {
core/ncore.go
+22
-5
@@ -591,7 +591,7 @@ func onlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo, cfg *iconfig.Conf
591
return namesys.NewNameSystem(rt, repo.Datastore(), cs), nil
592
}
593
594
-func ipnsRepublisher(lc fx.Lifecycle, cfg *iconfig.Config, namesys namesys.NameSystem, repo repo.Repo, privKey ic.PrivKey) error {
594
+func ipnsRepublisher(lc lcProcess, cfg *iconfig.Config, namesys namesys.NameSystem, repo repo.Repo, privKey ic.PrivKey) error {
595
repub := ipnsrp.NewRepublisher(namesys, repo.Datastore(), privKey, repo.Keystore())
596
597
if cfg.Ipns.RepublishPeriod != "" {
@@ -616,7 +616,7 @@ func ipnsRepublisher(lc fx.Lifecycle, cfg *iconfig.Config, namesys namesys.NameS
616
repub.RecordLifetime = d
617
}
618
619
- lcGoProc(lc, repub.Run)
619
+ lc.Run(repub.Run)
620
return nil
621
}
622
@@ -762,11 +762,18 @@ func lifecycleCtx(lc fx.Lifecycle) context.Context {
762
return ctx
763
}
764
765
-func lcGoProc(lc fx.Lifecycle, processFunc goprocess.ProcessFunc) {
765
+type lcProcess struct {
766
+ fx.In
767
+
768
+ LC fx.Lifecycle
769
+ Proc goprocess.Process
770
+}
771
+
772
+func (lp *lcProcess) Run(f goprocess.ProcessFunc) {
773
proc := make(chan goprocess.Process, 1)
767
- lc.Append(fx.Hook{
774
+ lp.LC.Append(fx.Hook{
775
OnStart: func(ctx context.Context) error {
769
- proc <- goprocess.Go(processFunc)
776
+ proc <- lp.Proc.Go(f)
777
return nil
778
},
779
OnStop: func(ctx context.Context) error {
@@ -775,6 +782,16 @@ func lcGoProc(lc fx.Lifecycle, processFunc goprocess.ProcessFunc) {
782
})
783
}
784
785
+func baseProcess(lc fx.Lifecycle) goprocess.Process {
786
+ p := goprocess.WithParent(goprocess.Background())
787
+ lc.Append(fx.Hook{
788
+ OnStop: func(_ context.Context) error {
789
+ return p.Close()
790
+ },
791
+ })
792
+ return p
793
+}
794
+
795
func setupSharding(cfg *iconfig.Config) {
796
// TEMP: setting global sharding switch here
797
uio.UseHAMTSharding = cfg.Experimental.ShardingEnabled
fuse/ipns/mount_unix.go
+1
-1
@@ -22,5 +22,5 @@ func Mount(ipfs *core.IpfsNode, ipnsmp, ipfsmp string) (mount.Mount, error) {
22
return nil, err
23
}
24
25
- return mount.NewMount(ipfs.Process(), fsys, ipnsmp, allow_other)
25
+ return mount.NewMount(ipfs.Process, fsys, ipnsmp, allow_other)
26
}
fuse/readonly/mount_unix.go
+1
-1
@@ -16,5 +16,5 @@ func Mount(ipfs *core.IpfsNode, mountpoint string) (mount.Mount, error) {
16
}
17
allow_other := cfg.Mounts.FuseAllowOther
18
fsys := NewFileSystem(ipfs)
19
- return mount.NewMount(ipfs.Process(), fsys, mountpoint, allow_other)
19
+ return mount.NewMount(ipfs.Process, fsys, mountpoint, allow_other)
20
}