feat(bitswap): synchronous close
Brian Tiger Chow committed
Jan 30, 2015 at 01:29 UTC
c114b04ae180a4471d5dc601c2ea5023ca0e8464
3 files changed
+47
-16
core/core.go
+6
-4
@@ -263,16 +263,18 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context) error {
263
func (n *IpfsNode) teardown() error {
264
// owned objects are closed in this teardown to ensure that they're closed
265
// regardless of which constructor was used to add them to the node.
266
- var closers []io.Closer
267
- addCloser := func(c io.Closer) {
266
+ closers := []io.Closer{
267
+ n.Blocks,
268
+ n.Exchange,
269
+ n.Repo,
270
+ }
271
+ addCloser := func(c io.Closer) { // use when field may be nil
272
if c != nil {
273
closers = append(closers, c)
274
}
275
}
276
277
addCloser(n.Bootstrapper)
274
- addCloser(n.Repo)
275
- addCloser(n.Blocks)
278
if dht, ok := n.Routing.(*dht.IpfsDHT); ok {
279
addCloser(dht)
280
}
exchange/bitswap/bitswap.go
+41
-10
@@ -9,6 +9,7 @@ import (
9
10
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
11
inflect "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/briantigerchow/inflect"
12
+ process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
13
14
blocks "github.com/jbenet/go-ipfs/blocks"
15
blockstore "github.com/jbenet/go-ipfs/blocks/blockstore"
@@ -52,28 +53,47 @@ var (
53
func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
54
bstore blockstore.Blockstore, nice bool) exchange.Interface {
55
56
+ // important to use provided parent context (since it may include important
57
+ // loggable data). It's probably not a good idea to allow bitswap to be
58
+ // coupled to the concerns of the IPFS daemon in this way.
59
+ //
60
+ // FIXME(btc) Now that bitswap manages itself using a process, it probably
61
+ // shouldn't accept a context anymore. Clients should probably use Close()
62
+ // exclusively. We should probably find another way to share logging data
63
ctx, cancelFunc := context.WithCancel(parent)
64
65
notif := notifications.New()
66
+ px := process.WithTeardown(func() error {
67
+ notif.Shutdown()
68
+ return nil
69
+ })
70
+
71
go func() {
59
- <-ctx.Done()
72
+ <-px.Closing() // process closes first
73
cancelFunc()
61
- notif.Shutdown()
74
+ }()
75
+ go func() {
76
+ <-ctx.Done() // parent cancelled first
77
+ px.Close()
78
}()
79
80
bs := &bitswap{
81
self: p,
82
blockstore: bstore,
67
- cancelFunc: cancelFunc,
83
notifications: notif,
69
- engine: decision.NewEngine(ctx, bstore),
84
+ engine: decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
85
network: network,
86
wantlist: wantlist.NewThreadSafe(),
87
batchRequests: make(chan []u.Key, sizeBatchRequestChan),
88
+ process: px,
89
}
90
network.SetDelegate(bs)
75
- go bs.clientWorker(ctx)
76
- go bs.taskWorker(ctx)
91
+ px.Go(func(px process.Process) {
92
+ bs.clientWorker(ctx)
93
+ })
94
+ px.Go(func(px process.Process) {
95
+ bs.taskWorker(ctx)
96
+ })
97
98
return bs
99
}
@@ -102,8 +122,7 @@ type bitswap struct {
122
123
wantlist *wantlist.ThreadSafe
124
105
- // cancelFunc signals cancellation to the bitswap event loop
106
- cancelFunc func()
125
+ process process.Process
126
}
127
128
// GetBlock attempts to retrieve a particular block from peers within the
@@ -149,6 +168,11 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err
168
// that lasts throughout the lifetime of the server)
169
func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
170
171
+ select {
172
+ case <-bs.process.Closing():
173
+ return nil, errors.New("bitswap is closed")
174
+ default:
175
+ }
176
promise := bs.notifications.Subscribe(ctx, keys...)
177
select {
178
case bs.batchRequests <- keys:
@@ -161,6 +185,11 @@ func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.
185
// HasBlock announces the existance of a block to this bitswap service. The
186
// service will potentially notify its peers.
187
func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
188
+ select {
189
+ case <-bs.process.Closing():
190
+ return errors.New("bitswap is closed")
191
+ default:
192
+ }
193
if err := bs.blockstore.Put(blk); err != nil {
194
return err
195
}
@@ -235,6 +264,7 @@ func (bs *bitswap) sendWantlistToProviders(ctx context.Context, entries []wantli
264
}
265
266
func (bs *bitswap) taskWorker(ctx context.Context) {
267
+ defer log.Info("bitswap task worker shutting down...")
268
for {
269
select {
270
case <-ctx.Done():
@@ -256,6 +286,8 @@ func (bs *bitswap) taskWorker(ctx context.Context) {
286
// TODO ensure only one active request per key
287
func (bs *bitswap) clientWorker(parent context.Context) {
288
289
+ defer log.Info("bitswap client worker shutting down...")
290
+
291
ctx, cancel := context.WithCancel(parent)
292
293
broadcastSignal := time.After(rebroadcastDelay.Get())
@@ -384,6 +416,5 @@ func (bs *bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage)
416
}
417
418
func (bs *bitswap) Close() error {
387
- bs.cancelFunc()
388
- return nil // to conform to Closer interface
419
+ return bs.process.Close()
420
}
exchange/bitswap/bitswap_test.go
-2
@@ -22,8 +22,6 @@ import (
22
const kNetworkDelay = 0 * time.Millisecond
23
24
func TestClose(t *testing.T) {
25
- // TODO
26
- t.Skip("TODO Bitswap's Close implementation is a WIP")
25
vnet := tn.VirtualNetwork(mockrouting.NewServer(), delay.Fixed(kNetworkDelay))
26
sesgen := NewTestSessionGenerator(vnet)
27
defer sesgen.Close()