blockstore: extract ARC cache from Bloom cache
it removes race condition that would happen during various calls License: MIT Signed-off-by: Jakub Sztandera <kubuxu@protonmail.ch>
Jakub Sztandera committed
Aug 2, 2016 at 01:10 UTC
4183902e4620bd24675555718497975da6863420
5 files changed
+206
-89
blocks/blockstore/arc_cache.go
new
+127
@@ -0,0 +1,127 @@
1
+package blockstore
2
+
3
+import (
4
+ "github.com/ipfs/go-ipfs/blocks"
5
+ key "github.com/ipfs/go-ipfs/blocks/key"
6
+ ds "gx/ipfs/QmTxLSvdhwg68WJimdS6icLPhZi28aTp6b7uihC2Yb47Xk/go-datastore"
7
+ lru "gx/ipfs/QmVYxfoJQiZijTgPNHCHgHELvQpbsJNTg6Crmc3dQkj3yy/golang-lru"
8
+ context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
9
+)
10
+
11
+type arccache struct {
12
+ arc *lru.ARCCache
13
+ blockstore Blockstore
14
+}
15
+
16
+func arcCached(bs Blockstore, lruSize int) (*arccache, error) {
17
+ arc, err := lru.NewARC(lruSize)
18
+ if err != nil {
19
+ return nil, err
20
+ }
21
+
22
+ return &arccache{arc: arc, blockstore: bs}, nil
23
+}
24
+
25
+func (b *arccache) DeleteBlock(k key.Key) error {
26
+ if has, ok := b.hasCached(k); ok && !has {
27
+ return ErrNotFound
28
+ }
29
+
30
+ b.arc.Remove(k) // Invalidate cache before deleting.
31
+ err := b.blockstore.DeleteBlock(k)
32
+ switch err {
33
+ case nil:
34
+ b.arc.Add(k, false)
35
+ case ds.ErrNotFound, ErrNotFound:
36
+ b.arc.Add(k, false)
37
+ default:
38
+ return err
39
+ }
40
+ return nil
41
+}
42
+
43
+// if ok == false has is inconclusive
44
+// if ok == true then has respons to question: is it contained
45
+func (b *arccache) hasCached(k key.Key) (has bool, ok bool) {
46
+ if k == "" {
47
+ // Return cache invalid so call to blockstore
48
+ // in case of invalid key is forwarded deeper
49
+ return false, false
50
+ }
51
+ h, ok := b.arc.Get(k)
52
+ if ok {
53
+ return h.(bool), ok
54
+ } else {
55
+ return false, false
56
+ }
57
+}
58
+
59
+func (b *arccache) Has(k key.Key) (bool, error) {
60
+ if has, ok := b.hasCached(k); ok {
61
+ return has, nil
62
+ }
63
+
64
+ res, err := b.blockstore.Has(k)
65
+ if err == nil {
66
+ b.arc.Add(k, res)
67
+ }
68
+ return res, err
69
+}
70
+
71
+func (b *arccache) Get(k key.Key) (blocks.Block, error) {
72
+ if has, ok := b.hasCached(k); ok && !has {
73
+ return nil, ErrNotFound
74
+ }
75
+
76
+ bl, err := b.blockstore.Get(k)
77
+ if bl == nil && err == ErrNotFound {
78
+ b.arc.Add(k, false)
79
+ } else if bl != nil {
80
+ b.arc.Add(k, true)
81
+ }
82
+ return bl, err
83
+}
84
+
85
+func (b *arccache) Put(bl blocks.Block) error {
86
+ if has, ok := b.hasCached(bl.Key()); ok && has {
87
+ return nil
88
+ }
89
+
90
+ err := b.blockstore.Put(bl)
91
+ if err == nil {
92
+ b.arc.Add(bl.Key(), true)
93
+ }
94
+ return err
95
+}
96
+
97
+func (b *arccache) PutMany(bs []blocks.Block) error {
98
+ var good []blocks.Block
99
+ for _, block := range bs {
100
+ if has, ok := b.hasCached(block.Key()); !ok || (ok && !has) {
101
+ good = append(good, block)
102
+ }
103
+ }
104
+ err := b.blockstore.PutMany(bs)
105
+ if err == nil {
106
+ for _, block := range bs {
107
+ b.arc.Add(block.Key(), true)
108
+ }
109
+ }
110
+ return err
111
+}
112
+
113
+func (b *arccache) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
114
+ return b.blockstore.AllKeysChan(ctx)
115
+}
116
+
117
+func (b *arccache) GCLock() Unlocker {
118
+ return b.blockstore.(GCBlockstore).GCLock()
119
+}
120
+
121
+func (b *arccache) PinLock() Unlocker {
122
+ return b.blockstore.(GCBlockstore).PinLock()
123
+}
124
+
125
+func (b *arccache) GCRequested() bool {
126
+ return b.blockstore.(GCBlockstore).GCRequested()
127
+}
blocks/blockstore/arc_cache_test.go
new
+67
@@ -0,0 +1,67 @@
1
+package blockstore
2
+
3
+import (
4
+ "github.com/ipfs/go-ipfs/blocks"
5
+ "testing"
6
+
7
+ ds "gx/ipfs/QmTxLSvdhwg68WJimdS6icLPhZi28aTp6b7uihC2Yb47Xk/go-datastore"
8
+ syncds "gx/ipfs/QmTxLSvdhwg68WJimdS6icLPhZi28aTp6b7uihC2Yb47Xk/go-datastore/sync"
9
+ context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
10
+)
11
+
12
+func testArcCached(bs GCBlockstore, ctx context.Context) (*arccache, error) {
13
+ if ctx == nil {
14
+ ctx = context.TODO()
15
+ }
16
+ opts := DefaultCacheOpts()
17
+ opts.HasBloomFilterSize = 0
18
+ opts.HasBloomFilterHashes = 0
19
+ bbs, err := CachedBlockstore(bs, ctx, opts)
20
+ if err == nil {
21
+ return bbs.(*arccache), nil
22
+ } else {
23
+ return nil, err
24
+ }
25
+}
26
+
27
+func TestRemoveCacheEntryOnDelete(t *testing.T) {
28
+ b := blocks.NewBlock([]byte("foo"))
29
+ cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
30
+ bs := NewBlockstore(syncds.MutexWrap(cd))
31
+ cachedbs, err := testArcCached(bs, nil)
32
+ if err != nil {
33
+ t.Fatal(err)
34
+ }
35
+ cachedbs.Put(b)
36
+
37
+ cd.Lock()
38
+ writeHitTheDatastore := false
39
+ cd.Unlock()
40
+
41
+ cd.SetFunc(func() {
42
+ writeHitTheDatastore = true
43
+ })
44
+
45
+ cachedbs.DeleteBlock(b.Key())
46
+ cachedbs.Put(b)
47
+ if !writeHitTheDatastore {
48
+ t.Fail()
49
+ }
50
+}
51
+
52
+func TestElideDuplicateWrite(t *testing.T) {
53
+ cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
54
+ bs := NewBlockstore(syncds.MutexWrap(cd))
55
+ cachedbs, err := testArcCached(bs, nil)
56
+ if err != nil {
57
+ t.Fatal(err)
58
+ }
59
+
60
+ b1 := blocks.NewBlock([]byte("foo"))
61
+
62
+ cachedbs.Put(b1)
63
+ cd.SetFunc(func() {
64
+ t.Fatal("write hit the datastore")
65
+ })
66
+ cachedbs.Put(b1)
67
+}
blocks/blockstore/bloom_cache.go
+6
-40
@@ -3,8 +3,6 @@ package blockstore
3
import (
4
"github.com/ipfs/go-ipfs/blocks"
5
key "github.com/ipfs/go-ipfs/blocks/key"
6
- ds "gx/ipfs/QmTxLSvdhwg68WJimdS6icLPhZi28aTp6b7uihC2Yb47Xk/go-datastore"
7
- lru "gx/ipfs/QmVYxfoJQiZijTgPNHCHgHELvQpbsJNTg6Crmc3dQkj3yy/golang-lru"
6
bloom "gx/ipfs/QmWQ2SJisXwcCLsUXLwYCKSfyExXjFRW2WbBH5sqCUnwX5/bbloom"
7
context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
8
@@ -13,16 +11,12 @@ import (
11
12
// bloomCached returns Blockstore that caches Has requests using Bloom filter
13
// Size is size of bloom filter in bytes
16
-func bloomCached(bs Blockstore, ctx context.Context, bloomSize, hashCount, lruSize int) (*bloomcache, error) {
14
+func bloomCached(bs Blockstore, ctx context.Context, bloomSize, hashCount int) (*bloomcache, error) {
15
bl, err := bloom.New(float64(bloomSize), float64(hashCount))
16
if err != nil {
17
return nil, err
18
}
21
- arc, err := lru.NewARC(lruSize)
22
- if err != nil {
23
- return nil, err
24
- }
25
- bc := &bloomcache{blockstore: bs, bloom: bl, arc: arc}
19
+ bc := &bloomcache{blockstore: bs, bloom: bl}
20
bc.Invalidate()
21
go bc.Rebuild(ctx)
22
@@ -33,7 +27,6 @@ type bloomcache struct {
27
bloom *bloom.Bloom
28
active int32
29
36
- arc *lru.ARCCache
30
// This chan is only used for testing to wait for bloom to enable
31
rebuildChan chan struct{}
32
blockstore Blockstore
@@ -84,17 +77,7 @@ func (b *bloomcache) DeleteBlock(k key.Key) error {
77
return ErrNotFound
78
}
79
87
- b.arc.Remove(k) // Invalidate cache before deleting.
88
- err := b.blockstore.DeleteBlock(k)
89
- switch err {
90
- case nil:
91
- b.arc.Add(k, false)
92
- case ds.ErrNotFound, ErrNotFound:
93
- b.arc.Add(k, false)
94
- default:
95
- return err
96
- }
97
- return nil
80
+ return b.blockstore.DeleteBlock(k)
81
}
82
83
// if ok == false has is inconclusive
@@ -111,12 +94,7 @@ func (b *bloomcache) hasCached(k key.Key) (has bool, ok bool) {
94
return false, true
95
}
96
}
114
- h, ok := b.arc.Get(k)
115
- if ok {
116
- return h.(bool), ok
117
- } else {
118
- return false, false
119
- }
97
+ return false, false
98
}
99
100
func (b *bloomcache) Has(k key.Key) (bool, error) {
@@ -124,11 +102,7 @@ func (b *bloomcache) Has(k key.Key) (bool, error) {
102
return has, nil
103
}
104
127
- res, err := b.blockstore.Has(k)
128
- if err == nil {
129
- b.arc.Add(k, res)
130
- }
131
- return res, err
105
+ return b.blockstore.Has(k)
106
}
107
108
func (b *bloomcache) Get(k key.Key) (blocks.Block, error) {
@@ -136,13 +110,7 @@ func (b *bloomcache) Get(k key.Key) (blocks.Block, error) {
110
return nil, ErrNotFound
111
}
112
139
- bl, err := b.blockstore.Get(k)
140
- if bl == nil && err == ErrNotFound {
141
- b.arc.Add(k, false)
142
- } else if bl != nil {
143
- b.arc.Add(k, true)
144
- }
145
- return bl, err
113
+ return b.blockstore.Get(k)
114
}
115
116
func (b *bloomcache) Put(bl blocks.Block) error {
@@ -153,7 +121,6 @@ func (b *bloomcache) Put(bl blocks.Block) error {
121
err := b.blockstore.Put(bl)
122
if err == nil {
123
b.bloom.AddTS([]byte(bl.Key()))
156
- b.arc.Add(bl.Key(), true)
124
}
125
return err
126
}
@@ -169,7 +136,6 @@ func (b *bloomcache) PutMany(bs []blocks.Block) error {
136
if err == nil {
137
for _, block := range bs {
138
b.bloom.AddTS([]byte(block.Key()))
172
- b.arc.Add(block.Key(), true)
139
}
140
}
141
return err
blocks/blockstore/bloom_cache_test.go
+2
-47
@@ -19,6 +19,7 @@ func testBloomCached(bs GCBlockstore, ctx context.Context) (*bloomcache, error)
19
ctx = context.TODO()
20
}
21
opts := DefaultCacheOpts()
22
+ opts.HasARCCacheSize = 0
23
bbs, err := CachedBlockstore(bs, ctx, opts)
24
if err == nil {
25
return bbs.(*bloomcache), nil
@@ -29,56 +30,10 @@ func testBloomCached(bs GCBlockstore, ctx context.Context) (*bloomcache, error)
30
31
func TestReturnsErrorWhenSizeNegative(t *testing.T) {
32
bs := NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore()))
32
- _, err := bloomCached(bs, context.TODO(), 100, 1, -1)
33
+ _, err := bloomCached(bs, context.TODO(), -1, 1)
34
if err == nil {
35
t.Fail()
36
}
36
- _, err = bloomCached(bs, context.TODO(), -1, 1, 100)
37
- if err == nil {
38
- t.Fail()
39
- }
40
-}
41
-
42
-func TestRemoveCacheEntryOnDelete(t *testing.T) {
43
- b := blocks.NewBlock([]byte("foo"))
44
- cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
45
- bs := NewBlockstore(syncds.MutexWrap(cd))
46
- cachedbs, err := testBloomCached(bs, nil)
47
- if err != nil {
48
- t.Fatal(err)
49
- }
50
- cachedbs.Put(b)
51
-
52
- cd.Lock()
53
- writeHitTheDatastore := false
54
- cd.Unlock()
55
-
56
- cd.SetFunc(func() {
57
- writeHitTheDatastore = true
58
- })
59
-
60
- cachedbs.DeleteBlock(b.Key())
61
- cachedbs.Put(b)
62
- if !writeHitTheDatastore {
63
- t.Fail()
64
- }
65
-}
66
-
67
-func TestElideDuplicateWrite(t *testing.T) {
68
- cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
69
- bs := NewBlockstore(syncds.MutexWrap(cd))
70
- cachedbs, err := testBloomCached(bs, nil)
71
- if err != nil {
72
- t.Fatal(err)
73
- }
74
-
75
- b1 := blocks.NewBlock([]byte("foo"))
76
-
77
- cachedbs.Put(b1)
78
- cd.SetFunc(func() {
79
- t.Fatal("write hit the datastore")
80
- })
81
- cachedbs.Put(b1)
37
}
38
func TestHasIsBloomCached(t *testing.T) {
39
cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
blocks/blockstore/caching.go
+4
-2
@@ -34,8 +34,10 @@ func CachedBlockstore(bs GCBlockstore,
34
return nil, errors.New("bloom filter hash count can't be 0 when there is size set")
35
}
36
if opts.HasBloomFilterSize != 0 {
37
- cbs, err = bloomCached(cbs, ctx, opts.HasBloomFilterSize, opts.HasBloomFilterHashes,
38
- opts.HasARCCacheSize)
37
+ cbs, err = bloomCached(cbs, ctx, opts.HasBloomFilterSize, opts.HasBloomFilterHashes)
38
+ }
39
+ if opts.HasARCCacheSize > 0 {
40
+ cbs, err = arcCached(cbs, opts.HasARCCacheSize)
41
}
42
43
return cbs, err