Change Process interface into object variable
License: MIT Signed-off-by: rht <rhtbot@gmail.com>
rht committed
Jun 18, 2015 at 17:17 UTC
007a12e7efb2b26eb79d335f2e8b968c4c541623
14 files changed
+106
-60
core/core.go
+30
-13
@@ -106,7 +106,8 @@ type IpfsNode struct {
106
107
IpnsFs *ipnsfs.Filesystem
108
109
- goprocess.Process
109
+ proc goprocess.Process
110
+ ctx context.Context
111
112
mode mode
113
}
@@ -121,23 +122,24 @@ type Mounts struct {
122
123
type ConfigOption func(ctx context.Context) (*IpfsNode, error)
124
124
-func NewIPFSNode(parent context.Context, option ConfigOption) (*IpfsNode, error) {
125
- procctx := goprocessctx.WithContext(parent)
126
- ctx := parent
125
+func NewIPFSNode(ctx context.Context, option ConfigOption) (*IpfsNode, error) {
126
+ node, err := option(ctx)
127
+ if err != nil {
128
+ return nil, err
129
+ }
130
+
131
+ proc := goprocessctx.WithContext(ctx)
132
+ proc.SetTeardown(node.teardown)
133
+ node.proc = proc
134
+ node.ctx = ctx
135
+
136
success := false // flip to true after all sub-system inits succeed
137
defer func() {
138
if !success {
130
- procctx.Close()
139
+ proc.Close()
140
}
141
}()
142
134
- node, err := option(ctx)
135
- if err != nil {
136
- return nil, err
137
- }
138
- node.Process = procctx
139
- ctxg.SetTeardown(node.teardown)
140
-
143
// Need to make sure it's perfectly clear 1) which variables are expected
144
// to be initialized at this point, and 2) which variables will be
145
// initialized after this point.
@@ -346,6 +348,21 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost
348
return nil
349
}
350
351
+// Process returns the Process object
352
+func (n *IpfsNode) Process() goprocess.Process {
353
+ return n.proc
354
+}
355
+
356
+// Close calls Close() on the Process object
357
+func (n *IpfsNode) Close() error {
358
+ return n.proc.Close()
359
+}
360
+
361
+// Context returns the IpfsNode context
362
+func (n *IpfsNode) Context() context.Context {
363
+ return n.ctx
364
+}
365
+
366
// teardown closes owned children. If any errors occur, this function returns
367
// the first error.
368
func (n *IpfsNode) teardown() error {
@@ -372,7 +389,7 @@ func (n *IpfsNode) teardown() error {
389
}
390
391
if dht, ok := n.Routing.(*dht.IpfsDHT); ok {
375
- closers = append(closers, dht)
392
+ closers = append(closers, dht.Process())
393
}
394
395
if n.PeerHost != nil {
core/corehttp/corehttp.go
+4
-6
@@ -12,6 +12,7 @@ import (
12
13
ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
14
manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
15
+ "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
16
core "github.com/ipfs/go-ipfs/core"
17
eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
18
)
@@ -78,20 +79,17 @@ func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error
79
var serverError error
80
serverExited := make(chan struct{})
81
81
- node.Children().Add(1)
82
- defer node.Children().Done()
83
-
84
- go func() {
82
+ node.Process().Go(func(p goprocess.Process) {
83
serverError = http.Serve(lis, handler)
84
close(serverExited)
87
- }()
85
+ })
86
87
// wait for server to exit.
88
select {
89
case <-serverExited:
90
91
// if node being closed before server exits, close server
94
- case <-node.Closing():
92
+ case <-node.Process().Closing():
93
log.Infof("server at %s terminating...", addr)
94
95
lis.Close()
core/coreunix/cat.go
+2
-2
@@ -10,9 +10,9 @@ import (
10
11
func Cat(n *core.IpfsNode, pstr string) (io.Reader, error) {
12
p := path.FromString(pstr)
13
- dagNode, err := n.Resolver.ResolvePath(n.ContextGroup.Context(), p)
13
+ dagNode, err := n.Resolver.ResolvePath(n.Context(), p)
14
if err != nil {
15
return nil, err
16
}
17
- return uio.NewDagReader(n.ContextGroup.Context(), dagNode, n.DAG)
17
+ return uio.NewDagReader(n.Context(), dagNode, n.DAG)
18
}
core/mock/mock.go
+16
-11
@@ -3,7 +3,6 @@ package coremock
3
import (
4
"github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
5
syncds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
6
- goprocessctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
6
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
7
"github.com/ipfs/go-ipfs/blocks/blockstore"
8
blockservice "github.com/ipfs/go-ipfs/blockservice"
@@ -16,6 +15,7 @@ import (
15
path "github.com/ipfs/go-ipfs/path"
16
pin "github.com/ipfs/go-ipfs/pin"
17
"github.com/ipfs/go-ipfs/repo"
18
+ config "github.com/ipfs/go-ipfs/repo/config"
19
offrt "github.com/ipfs/go-ipfs/routing/offline"
20
ds2 "github.com/ipfs/go-ipfs/util/datastore2"
21
testutil "github.com/ipfs/go-ipfs/util/testutil"
@@ -28,33 +28,38 @@ import (
28
// NewMockNode constructs an IpfsNode for use in tests.
29
func NewMockNode() (*core.IpfsNode, error) {
30
ctx := context.TODO()
31
- nd := new(core.IpfsNode)
31
32
// Generate Identity
33
ident, err := testutil.RandIdentity()
34
if err != nil {
35
return nil, err
36
}
38
-
37
p := ident.ID()
40
- nd.Identity = p
38
+
39
+ c := config.Config{
40
+ Identity: config.Identity{
41
+ PeerID: p.String(),
42
+ },
43
+ }
44
+
45
+ nd, err := core.Offline(&repo.Mock{
46
+ C: c,
47
+ D: ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore())),
48
+ })(ctx)
49
+ if err != nil {
50
+ return nil, err
51
+ }
52
+
53
nd.PrivateKey = ident.PrivateKey()
54
nd.Peerstore = peer.NewPeerstore()
55
nd.Peerstore.AddPrivKey(p, ident.PrivateKey())
56
nd.Peerstore.AddPubKey(p, ident.PublicKey())
45
- nd.Process = goprocessctx.WithContext(ctx)
57
58
nd.PeerHost, err = mocknet.New(ctx).AddPeer(ident.PrivateKey(), ident.Address()) // effectively offline
59
if err != nil {
60
return nil, err
61
}
62
52
- // Temp Datastore
53
- nd.Repo = &repo.Mock{
54
- // TODO C: conf,
55
- D: ds2.CloserWrap(syncds.MutexWrap(datastore.NewMapDatastore())),
56
- }
57
-
63
// Routing
64
nd.Routing = offrt.NewOfflineRouter(nd.Repo.Datastore(), nd.PrivateKey)
65
fuse/ipns/mount_unix.go
+1
-1
@@ -18,5 +18,5 @@ func Mount(ipfs *core.IpfsNode, ipnsmp, ipfsmp string) (mount.Mount, error) {
18
return nil, err
19
}
20
21
- return mount.NewMount(ipfs, fsys, ipnsmp, allow_other)
21
+ return mount.NewMount(ipfs.Process(), fsys, ipnsmp, allow_other)
22
}
fuse/readonly/mount_unix.go
+1
-1
@@ -13,5 +13,5 @@ func Mount(ipfs *core.IpfsNode, mountpoint string) (mount.Mount, error) {
13
cfg := ipfs.Repo.Config()
14
allow_other := cfg.Mounts.FuseAllowOther
15
fsys := NewFileSystem(ipfs)
16
- return mount.NewMount(ipfs, fsys, mountpoint, allow_other)
16
+ return mount.NewMount(ipfs.Process(), fsys, mountpoint, allow_other)
17
}
p2p/net/mock/mock_net.go
+6
-4
@@ -33,6 +33,7 @@ type mocknet struct {
33
linkDefaults LinkOptions
34
35
proc goprocess.Process // for Context closing
36
+ ctx context.Context
37
sync.RWMutex
38
}
39
@@ -42,6 +43,7 @@ func New(ctx context.Context) Mocknet {
43
hosts: map[peer.ID]*bhost.BasicHost{},
44
links: map[peer.ID]map[peer.ID]map[*link]struct{}{},
45
proc: goprocessctx.WithContext(ctx),
46
+ ctx: ctx,
47
}
48
}
49
@@ -62,7 +64,7 @@ func (mn *mocknet) GenPeer() (host.Host, error) {
64
}
65
66
func (mn *mocknet) AddPeer(k ic.PrivKey, a ma.Multiaddr) (host.Host, error) {
65
- n, err := newPeernet(mn.cg.Context(), mn, k, a)
67
+ n, err := newPeernet(mn.ctx, mn, k, a)
68
if err != nil {
69
return nil, err
70
}
@@ -70,7 +72,7 @@ func (mn *mocknet) AddPeer(k ic.PrivKey, a ma.Multiaddr) (host.Host, error) {
72
h := bhost.New(n)
73
log.Debugf("mocknet added listen addr for peer: %s -- %s", n.LocalPeer(), a)
74
73
- mn.cg.AddChild(n.cg)
75
+ mn.proc.AddChild(n.proc)
76
77
mn.Lock()
78
mn.nets[n.peer] = n
@@ -298,11 +300,11 @@ func (mn *mocknet) ConnectAll() error {
300
}
301
302
func (mn *mocknet) ConnectPeers(a, b peer.ID) (inet.Conn, error) {
301
- return mn.Net(a).DialPeer(mn.cg.Context(), b)
303
+ return mn.Net(a).DialPeer(mn.ctx, b)
304
}
305
306
func (mn *mocknet) ConnectNets(a, b inet.Network) (inet.Conn, error) {
305
- return a.DialPeer(mn.cg.Context(), b.LocalPeer())
307
+ return a.DialPeer(mn.ctx, b.LocalPeer())
308
}
309
310
func (mn *mocknet) DisconnectPeers(p1, p2 peer.ID) error {
p2p/net/mock/mock_peernet.go
+2
-2
@@ -66,7 +66,7 @@ func newPeernet(ctx context.Context, m *mocknet, k ic.PrivKey,
66
notifs: make(map[inet.Notifiee]struct{}),
67
}
68
69
- n.cg.SetTeardown(n.teardown)
69
+ n.proc.SetTeardown(n.teardown)
70
return n, nil
71
}
72
@@ -94,7 +94,7 @@ func (pn *peernet) allConns() []*conn {
94
95
// Close calls the ContextCloser func
96
func (pn *peernet) Close() error {
97
- return pn.cg.Close()
97
+ return pn.proc.Close()
98
}
99
100
func (pn *peernet) Peerstore() peer.Peerstore {
p2p/net/swarm/swarm.go
+7
@@ -65,6 +65,7 @@ type Swarm struct {
65
Filters *filter.Filters
66
67
proc goprocess.Process
68
+ ctx context.Context
69
bwc metrics.Reporter
70
}
71
@@ -82,6 +83,7 @@ func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr,
83
local: local,
84
peers: peers,
85
proc: goprocessctx.WithContext(ctx),
86
+ ctx: ctx,
87
dialT: DialTimeout,
88
notifs: make(map[inet.Notifiee]ps.Notifiee),
89
bwc: bwc,
@@ -137,6 +139,11 @@ func (s *Swarm) Process() goprocess.Process {
139
return s.proc
140
}
141
142
+// Context returns the context of the swarm
143
+func (s *Swarm) Context() context.Context {
144
+ return s.ctx
145
+}
146
+
147
// Close stops the Swarm.
148
func (s *Swarm) Close() error {
149
return s.proc.Close()
p2p/net/swarm/swarm_listen.go
+2
-2
@@ -64,7 +64,7 @@ func (s *Swarm) setupListener(maddr ma.Multiaddr) error {
64
log.Warning("Listener not given PrivateKey, so WILL NOT SECURE conns.")
65
}
66
log.Debugf("Swarm Listening at %s", maddr)
67
- list, err := conn.Listen(s.cg.Context(), maddr, s.local, sk)
67
+ list, err := conn.Listen(s.Context(), maddr, s.local, sk)
68
if err != nil {
69
return err
70
}
@@ -112,7 +112,7 @@ func (s *Swarm) setupListener(maddr ma.Multiaddr) error {
112
return
113
}
114
}
115
- }(s.cg.Context(), sl)
115
+ }(s.Context(), sl)
116
117
return nil
118
}
routing/dht/dht.go
+28
-11
@@ -58,8 +58,8 @@ type IpfsDHT struct {
58
59
Validator record.Validator // record validator funcs
60
61
- Context context.Context
62
- goprocess.Process
61
+ ctx context.Context
62
+ proc goprocess.Process
63
}
64
65
// NewDHT creates a new DHT object with the given peer as the 'local' host
@@ -73,18 +73,18 @@ func NewDHT(ctx context.Context, h host.Host, dstore ds.ThreadSafeDatastore) *Ip
73
// register for network notifs.
74
dht.host.Network().Notify((*netNotifiee)(dht))
75
76
- procctx = goprocessctx.WithContext(ctx)
77
- procctx.SetTeardown(func() error {
76
+ proc := goprocessctx.WithContext(ctx)
77
+ proc.SetTeardown(func() error {
78
// remove ourselves from network notifs.
79
dht.host.Network().StopNotify((*netNotifiee)(dht))
80
return nil
81
})
82
- dht.Process = procctx
83
- dht.Context = ctx
82
+ dht.proc = proc
83
+ dht.ctx = ctx
84
85
h.SetStreamHandler(ProtocolDHT, dht.handleNewStream)
86
- dht.providers = NewProviderManager(dht.Context, dht.self)
87
- dht.AddChild(dht.providers)
86
+ dht.providers = NewProviderManager(dht.ctx, dht.self)
87
+ dht.proc.AddChild(dht.providers.proc)
88
89
dht.routingTable = kb.NewRoutingTable(20, kb.ConvertPeerID(dht.self), time.Minute, dht.peerstore)
90
dht.birth = time.Now()
@@ -93,7 +93,9 @@ func NewDHT(ctx context.Context, h host.Host, dstore ds.ThreadSafeDatastore) *Ip
93
dht.Validator["pk"] = record.PublicKeyValidator
94
95
if doPinging {
96
- dht.Go(func() { dht.PingRoutine(time.Second * 10) })
96
+ dht.proc.Go(func(p goprocess.Process) {
97
+ dht.PingRoutine(time.Second * 10)
98
+ })
99
}
100
return dht
101
}
@@ -360,15 +362,30 @@ func (dht *IpfsDHT) PingRoutine(t time.Duration) {
362
rand.Read(id)
363
peers := dht.routingTable.NearestPeers(kb.ConvertKey(key.Key(id)), 5)
364
for _, p := range peers {
363
- ctx, cancel := context.WithTimeout(dht.Context, time.Second*5)
365
+ ctx, cancel := context.WithTimeout(dht.Context(), time.Second*5)
366
_, err := dht.Ping(ctx, p)
367
if err != nil {
368
log.Debugf("Ping error: %s", err)
369
}
370
cancel()
371
}
370
- case <-dht.Closing():
372
+ case <-dht.proc.Closing():
373
return
374
}
375
}
376
}
377
+
378
+// Context return dht's context
379
+func (dht *IpfsDHT) Context() context.Context {
380
+ return dht.ctx
381
+}
382
+
383
+// Process return dht's process
384
+func (dht *IpfsDHT) Process() goprocess.Process {
385
+ return dht.proc
386
+}
387
+
388
+// Close calls Process Close
389
+func (dht *IpfsDHT) Close() error {
390
+ return dht.proc.Close()
391
+}
routing/dht/notif.go
+2
-2
@@ -16,7 +16,7 @@ func (nn *netNotifiee) DHT() *IpfsDHT {
16
func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
17
dht := nn.DHT()
18
select {
19
- case <-dht.Closing():
19
+ case <-dht.Process().Closing():
20
return
21
default:
22
}
@@ -26,7 +26,7 @@ func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
26
func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
27
dht := nn.DHT()
28
select {
29
- case <-dht.Closing():
29
+ case <-dht.Process().Closing():
30
return
31
default:
32
}
routing/dht/providers.go
+4
-4
@@ -22,7 +22,7 @@ type ProviderManager struct {
22
newprovs chan *addProv
23
getprovs chan *getProv
24
period time.Duration
25
- goprocess.Process
25
+ proc goprocess.Process
26
}
27
28
type providerSet struct {
@@ -47,8 +47,8 @@ func NewProviderManager(ctx context.Context, local peer.ID) *ProviderManager {
47
pm.providers = make(map[key.Key]*providerSet)
48
pm.getlocal = make(chan chan []key.Key)
49
pm.local = make(map[key.Key]struct{})
50
- pm.Process = goprocessctx.WithContext(ctx)
51
- pm.Go(pm.run)
50
+ pm.proc = goprocessctx.WithContext(ctx)
51
+ pm.proc.Go(func(p goprocess.Process) { pm.run() })
52
53
return pm
54
}
@@ -97,7 +97,7 @@ func (pm *ProviderManager) run() {
97
provs.providers = filtered
98
}
99
100
- case <-pm.Closing():
100
+ case <-pm.proc.Closing():
101
return
102
}
103
}
routing/dht/providers_test.go
+1
-1
@@ -19,5 +19,5 @@ func TestProviderManager(t *testing.T) {
19
if len(resp) != 1 {
20
t.Fatal("Could not retrieve provider.")
21
}
22
- p.Close()
22
+ p.proc.Close()
23
}