fix(notifications) prevent deadlock when context cancelled early
+ test(notifications) cc @whyrusleeping @jbenet License: MIT Signed-off-by: Brian Tiger Chow <brian@perfmode.com>
Brian Tiger Chow committed
Nov 26, 2014 at 12:06 UTC
bb0b5f7c0b682a36fedb6fe4f0d1e82797d877a3
2 files changed
+34
-4
exchange/bitswap/notifications/notifications.go
+4
-4
@@ -39,15 +39,15 @@ func (ps *impl) Shutdown() {
39
func (ps *impl) Subscribe(ctx context.Context, keys ...u.Key) <-chan *blocks.Block {
40
41
blocksCh := make(chan *blocks.Block, len(keys))
42
+ valuesCh := make(chan interface{}, len(keys)) // provide our own channel to control buffer, prevent blocking
43
if len(keys) == 0 {
44
close(blocksCh)
45
return blocksCh
46
}
46
- valuesCh := ps.wrapped.SubOnceEach(toStrings(keys)...)
47
+ ps.wrapped.AddSubOnceEach(valuesCh, toStrings(keys)...)
48
go func() {
48
- defer func() {
49
- close(blocksCh)
50
- }()
49
+ defer close(blocksCh)
50
+ defer ps.wrapped.Unsub(valuesCh) // with a len(keys) buffer, this is an optimization
51
for {
52
select {
53
case <-ctx.Done():
exchange/bitswap/notifications/notifications_test.go
+30
@@ -7,6 +7,8 @@ import (
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
blocks "github.com/jbenet/go-ipfs/blocks"
10
+ blocksutil "github.com/jbenet/go-ipfs/blocks/blocksutil"
11
+ "github.com/jbenet/go-ipfs/util"
12
)
13
14
func TestDuplicates(t *testing.T) {
@@ -96,6 +98,34 @@ func TestCarryOnWhenDeadlineExpires(t *testing.T) {
98
assertBlockChannelNil(t, blockChannel)
99
}
100
101
+func TestDoesNotDeadLockIfContextCancelledBeforePublish(t *testing.T) {
102
+
103
+ g := blocksutil.NewBlockGenerator()
104
+ ctx, cancel := context.WithCancel(context.Background())
105
+ n := New()
106
+ defer n.Shutdown()
107
+
108
+ t.Log("generate a large number of blocks. exceed default buffer")
109
+ bs := g.Blocks(1000)
110
+ ks := func() []util.Key {
111
+ var keys []util.Key
112
+ for _, b := range bs {
113
+ keys = append(keys, b.Key())
114
+ }
115
+ return keys
116
+ }()
117
+
118
+ _ = n.Subscribe(ctx, ks...) // ignore received channel
119
+
120
+ t.Log("cancel context before any blocks published")
121
+ cancel()
122
+ for _, b := range bs {
123
+ n.Publish(b)
124
+ }
125
+
126
+ t.Log("publishing the large number of blocks to the ignored channel must not deadlock")
127
+}
128
+
129
func assertBlockChannelNil(t *testing.T, blockChannel <-chan *blocks.Block) {
130
_, ok := <-blockChannel
131
if ok {