blocks/blockstore: Add bloom filter
Replace write_cache with bloom_cache Improve ARC caching Fix small issue in case of AllKeysChan fails deps: Update go-datastore blocks/blockstore: Invalidate ARC cache before deletin block deps: Update go-datastore License: MIT Signed-off-by: Jakub Sztandera <kubuxu@protonmail.ch>
Jakub Sztandera committed
Jun 21, 2016 at 21:05 UTC
5d83d89f360e02f845d1896e5d336c10d5469a4e
6 files changed
+227
-88
blocks/blockstore/bloom_cache.go
new
+175
@@ -0,0 +1,175 @@
1
+package blockstore
2
+
3
+import (
4
+ "github.com/ipfs/go-ipfs/blocks"
5
+ key "github.com/ipfs/go-ipfs/blocks/key"
6
+ lru "gx/ipfs/QmVYxfoJQiZijTgPNHCHgHELvQpbsJNTg6Crmc3dQkj3yy/golang-lru"
7
+ bloom "gx/ipfs/QmWQ2SJisXwcCLsUXLwYCKSfyExXjFRW2WbBH5sqCUnwX5/bbloom"
8
+ context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
9
+ ds "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore"
10
+)
11
+
12
+// BloomCached returns Blockstore that caches Has requests using Bloom filter
13
+// Size is size of bloom filter in bytes
14
+func BloomCached(bs Blockstore, bloomSize, lruSize int) (*bloomcache, error) {
15
+ bl, err := bloom.New(float64(bloomSize), float64(7))
16
+ if err != nil {
17
+ return nil, err
18
+ }
19
+ arc, err := lru.NewARC(lruSize)
20
+ if err != nil {
21
+ return nil, err
22
+ }
23
+ bc := &bloomcache{blockstore: bs, bloom: bl, arc: arc}
24
+ bc.Invalidate()
25
+ go bc.Rebuild()
26
+
27
+ return bc, nil
28
+}
29
+
30
+type bloomcache struct {
31
+ bloom *bloom.Bloom
32
+ active bool
33
+
34
+ arc *lru.ARCCache
35
+ // This chan is only used for testing to wait for bloom to enable
36
+ rebuildChan chan struct{}
37
+ blockstore Blockstore
38
+
39
+ // Statistics
40
+ hits uint64
41
+ misses uint64
42
+}
43
+
44
+func (b *bloomcache) Invalidate() {
45
+ b.rebuildChan = make(chan struct{})
46
+ b.active = false
47
+}
48
+
49
+func (b *bloomcache) BloomActive() bool {
50
+ return b.active
51
+}
52
+
53
+func (b *bloomcache) Rebuild() {
54
+ ctx := context.TODO()
55
+ evt := log.EventBegin(ctx, "bloomcache.Rebuild")
56
+ defer evt.Done()
57
+
58
+ ch, err := b.blockstore.AllKeysChan(ctx)
59
+ if err != nil {
60
+ log.Errorf("AllKeysChan failed in bloomcache rebuild with: %v", err)
61
+ return
62
+ }
63
+ for key := range ch {
64
+ b.bloom.AddTS([]byte(key)) // Use binary key, the more compact the better
65
+ }
66
+ close(b.rebuildChan)
67
+ b.active = true
68
+}
69
+
70
+func (b *bloomcache) DeleteBlock(k key.Key) error {
71
+ if has, ok := b.hasCached(k); ok && !has {
72
+ return ErrNotFound
73
+ }
74
+
75
+ b.arc.Remove(k) // Invalidate cache before deleting.
76
+ err := b.blockstore.DeleteBlock(k)
77
+ if err == nil {
78
+ b.arc.Add(k, false)
79
+ } else if err == ds.ErrNotFound || err == ErrNotFound {
80
+ b.arc.Add(k, false)
81
+ return ErrNotFound
82
+ }
83
+ return err
84
+}
85
+
86
+// if ok == false has is inconclusive
87
+// if ok == true then has respons to question: is it contained
88
+func (b *bloomcache) hasCached(k key.Key) (has bool, ok bool) {
89
+ if k == "" {
90
+ return true, true
91
+ }
92
+ if b.active {
93
+ blr := b.bloom.HasTS([]byte(k))
94
+ if blr == false { // not contained in bloom is only conclusive answer bloom gives
95
+ return blr, true
96
+ }
97
+ }
98
+ h, ok := b.arc.Get(k)
99
+ if ok {
100
+ return h.(bool), ok
101
+ } else {
102
+ return false, ok
103
+ }
104
+}
105
+
106
+func (b *bloomcache) Has(k key.Key) (bool, error) {
107
+ if has, ok := b.hasCached(k); ok {
108
+ return has, nil
109
+ }
110
+
111
+ res, err := b.blockstore.Has(k)
112
+ if err == nil {
113
+ b.arc.Add(k, res)
114
+ }
115
+ return res, err
116
+}
117
+
118
+func (b *bloomcache) Get(k key.Key) (blocks.Block, error) {
119
+ if has, ok := b.hasCached(k); ok && !has {
120
+ return nil, ErrNotFound
121
+ }
122
+
123
+ bl, err := b.blockstore.Get(k)
124
+ if bl == nil && err == ErrNotFound {
125
+ b.arc.Add(k, false)
126
+ } else if bl != nil {
127
+ b.arc.Add(k, true)
128
+ }
129
+ return bl, err
130
+}
131
+
132
+func (b *bloomcache) Put(bl blocks.Block) error {
133
+ if has, ok := b.hasCached(bl.Key()); ok && has {
134
+ return nil
135
+ }
136
+
137
+ err := b.blockstore.Put(bl)
138
+ if err == nil {
139
+ b.bloom.AddTS([]byte(bl.Key()))
140
+ b.arc.Add(bl.Key(), true)
141
+ }
142
+ return err
143
+}
144
+
145
+func (b *bloomcache) PutMany(bs []blocks.Block) error {
146
+ var good []blocks.Block
147
+ for _, block := range bs {
148
+ if has, ok := b.hasCached(block.Key()); !ok || (ok && !has) {
149
+ good = append(good, block)
150
+ }
151
+ }
152
+ err := b.blockstore.PutMany(bs)
153
+ if err == nil {
154
+ for _, block := range bs {
155
+ b.bloom.AddTS([]byte(block.Key()))
156
+ }
157
+ }
158
+ return err
159
+}
160
+
161
+func (b *bloomcache) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
162
+ return b.blockstore.AllKeysChan(ctx)
163
+}
164
+
165
+func (b *bloomcache) GCLock() Unlocker {
166
+ return b.blockstore.(GCBlockstore).GCLock()
167
+}
168
+
169
+func (b *bloomcache) PinLock() Unlocker {
170
+ return b.blockstore.(GCBlockstore).PinLock()
171
+}
172
+
173
+func (b *bloomcache) GCRequested() bool {
174
+ return b.blockstore.(GCBlockstore).GCRequested()
175
+}
blocks/blockstore/bloom_cache_test.go
renamed
+43
-8
@@ -1,28 +1,32 @@
1
package blockstore
2
3
import (
4
- "testing"
5
-
4
+ "fmt"
5
"github.com/ipfs/go-ipfs/blocks"
6
ds "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore"
7
dsq "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore/query"
8
syncds "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore/sync"
9
+ "testing"
10
+ "time"
11
)
12
13
func TestReturnsErrorWhenSizeNegative(t *testing.T) {
14
bs := NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore()))
14
- _, err := WriteCached(bs, -1)
15
- if err != nil {
16
- return
15
+ _, err := BloomCached(bs, 100, -1)
16
+ if err == nil {
17
+ t.Fail()
18
+ }
19
+ _, err = BloomCached(bs, -1, 100)
20
+ if err == nil {
21
+ t.Fail()
22
}
18
- t.Fail()
23
}
24
25
func TestRemoveCacheEntryOnDelete(t *testing.T) {
26
b := blocks.NewBlock([]byte("foo"))
27
cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
28
bs := NewBlockstore(syncds.MutexWrap(cd))
25
- cachedbs, err := WriteCached(bs, 1)
29
+ cachedbs, err := BloomCached(bs, 1, 1)
30
if err != nil {
31
t.Fatal(err)
32
}
@@ -43,7 +47,7 @@ func TestRemoveCacheEntryOnDelete(t *testing.T) {
47
func TestElideDuplicateWrite(t *testing.T) {
48
cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
49
bs := NewBlockstore(syncds.MutexWrap(cd))
46
- cachedbs, err := WriteCached(bs, 1)
50
+ cachedbs, err := BloomCached(bs, 1, 1)
51
if err != nil {
52
t.Fatal(err)
53
}
@@ -56,6 +60,37 @@ func TestElideDuplicateWrite(t *testing.T) {
60
})
61
cachedbs.Put(b1)
62
}
63
+func TestHasIsBloomCached(t *testing.T) {
64
+ cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()}
65
+ bs := NewBlockstore(syncds.MutexWrap(cd))
66
+
67
+ for i := 0; i < 1000; i++ {
68
+ bs.Put(blocks.NewBlock([]byte(fmt.Sprintf("data: %d", i))))
69
+ }
70
+ cachedbs, err := BloomCached(bs, 256*1024, 128)
71
+ if err != nil {
72
+ t.Fatal(err)
73
+ }
74
+
75
+ select {
76
+ case <-cachedbs.rebuildChan:
77
+ case <-time.After(1 * time.Second):
78
+ t.Fatalf("Timeout wating for rebuild: %d", cachedbs.bloom.ElementsAdded())
79
+ }
80
+
81
+ cacheFails := 0
82
+ cd.SetFunc(func() {
83
+ cacheFails++
84
+ })
85
+
86
+ for i := 0; i < 1000; i++ {
87
+ cachedbs.Has(blocks.NewBlock([]byte(fmt.Sprintf("data: %d", i+2000))).Key())
88
+ }
89
+
90
+ if float64(cacheFails)/float64(1000) > float64(0.05) {
91
+ t.Fatal("Bloom filter has cache miss rate of more than 5%")
92
+ }
93
+}
94
95
type callbackDatastore struct {
96
f func()
blocks/blockstore/write_cache.go
deleted
-78
@@ -1,78 +0,0 @@
1
-package blockstore
2
-
3
-import (
4
- "github.com/ipfs/go-ipfs/blocks"
5
- key "github.com/ipfs/go-ipfs/blocks/key"
6
- "gx/ipfs/QmVYxfoJQiZijTgPNHCHgHELvQpbsJNTg6Crmc3dQkj3yy/golang-lru"
7
- context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
8
-)
9
-
10
-// WriteCached returns a blockstore that caches up to |size| unique writes (bs.Put).
11
-func WriteCached(bs Blockstore, size int) (*writecache, error) {
12
- c, err := lru.New(size)
13
- if err != nil {
14
- return nil, err
15
- }
16
- return &writecache{blockstore: bs, cache: c}, nil
17
-}
18
-
19
-type writecache struct {
20
- cache *lru.Cache // pointer b/c Cache contains a Mutex as value (complicates copying)
21
- blockstore Blockstore
22
-}
23
-
24
-func (w *writecache) DeleteBlock(k key.Key) error {
25
- defer log.EventBegin(context.TODO(), "writecache.BlockRemoved", &k).Done()
26
- w.cache.Remove(k)
27
- return w.blockstore.DeleteBlock(k)
28
-}
29
-
30
-func (w *writecache) Has(k key.Key) (bool, error) {
31
- if _, ok := w.cache.Get(k); ok {
32
- return true, nil
33
- }
34
- return w.blockstore.Has(k)
35
-}
36
-
37
-func (w *writecache) Get(k key.Key) (blocks.Block, error) {
38
- return w.blockstore.Get(k)
39
-}
40
-
41
-func (w *writecache) Put(b blocks.Block) error {
42
- k := b.Key()
43
- if _, ok := w.cache.Get(k); ok {
44
- return nil
45
- }
46
- defer log.EventBegin(context.TODO(), "writecache.BlockAdded", &k).Done()
47
-
48
- w.cache.Add(b.Key(), struct{}{})
49
- return w.blockstore.Put(b)
50
-}
51
-
52
-func (w *writecache) PutMany(bs []blocks.Block) error {
53
- var good []blocks.Block
54
- for _, b := range bs {
55
- if _, ok := w.cache.Get(b.Key()); !ok {
56
- good = append(good, b)
57
- k := b.Key()
58
- defer log.EventBegin(context.TODO(), "writecache.BlockAdded", &k).Done()
59
- }
60
- }
61
- return w.blockstore.PutMany(good)
62
-}
63
-
64
-func (w *writecache) AllKeysChan(ctx context.Context) (<-chan key.Key, error) {
65
- return w.blockstore.AllKeysChan(ctx)
66
-}
67
-
68
-func (w *writecache) GCLock() Unlocker {
69
- return w.blockstore.(GCBlockstore).GCLock()
70
-}
71
-
72
-func (w *writecache) PinLock() Unlocker {
73
- return w.blockstore.(GCBlockstore).PinLock()
74
-}
75
-
76
-func (w *writecache) GCRequested() bool {
77
- return w.blockstore.(GCBlockstore).GCRequested()
78
-}
core/builder.go
+1
-1
@@ -131,7 +131,7 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
131
132
var err error
133
bs := bstore.NewBlockstore(n.Repo.Datastore())
134
- n.Blockstore, err = bstore.WriteCached(bs, kSizeBlockstoreWriteCache)
134
+ n.Blockstore, err = bstore.BloomCached(bs, 256*1024, kSizeBlockstoreWriteCache)
135
if err != nil {
136
return err
137
}
exchange/bitswap/testutils.go
+2
-1
@@ -87,12 +87,13 @@ func (i *Instance) SetBlockstoreLatency(t time.Duration) time.Duration {
87
// just a much better idea.
88
func Session(ctx context.Context, net tn.Network, p testutil.Identity) Instance {
89
bsdelay := delay.Fixed(0)
90
+ const bloomSize = 512
91
const writeCacheElems = 100
92
93
adapter := net.Adapter(p)
94
dstore := ds_sync.MutexWrap(datastore2.WithDelay(ds.NewMapDatastore(), bsdelay))
95
95
- bstore, err := blockstore.WriteCached(blockstore.NewBlockstore(ds_sync.MutexWrap(dstore)), writeCacheElems)
96
+ bstore, err := blockstore.BloomCached(blockstore.NewBlockstore(ds_sync.MutexWrap(dstore)), bloomSize, writeCacheElems)
97
if err != nil {
98
panic(err.Error()) // FIXME perhaps change signature and return error.
99
}
package.json
+6
@@ -177,6 +177,12 @@
177
"hash": "Qmb1DA2A9LS2wR4FFweB4uEDomFsdmnw1VLawLE1yQzudj",
178
"name": "base32",
179
"version": "0.0.0"
180
+ },
181
+ {
182
+ "author": "kubuxu",
183
+ "hash": "QmWQ2SJisXwcCLsUXLwYCKSfyExXjFRW2WbBH5sqCUnwX5",
184
+ "name": "bbloom",
185
+ "version": "0.0.2"
186
}
187
],
188
"gxVersion": "0.4.0",