Separate out the G.C. Locking from the Blockstore interface.
Factored out of #3257 (Add support for multiple blockstores). License: MIT Signed-off-by: Kevin Atkinson <k@kevina.org>
Kevin Atkinson committed
Nov 2, 2016 at 21:56 UTC
ffe9d7dae447e239c5d61478b9e0cd3ef7cf216c
7 files changed
+39
-15
blocks/blockstore/arc_cache_test.go
+1
-1
@@ -13,7 +13,7 @@ import (
13
14
var exampleBlock = blocks.NewBlock([]byte("foo"))
15
16
-func testArcCached(bs GCBlockstore, ctx context.Context) (*arccache, error) {
16
+func testArcCached(bs Blockstore, ctx context.Context) (*arccache, error) {
17
if ctx == nil {
18
ctx = context.TODO()
19
}
blocks/blockstore/blockstore.go
+28
-6
@@ -39,9 +39,7 @@ type Blockstore interface {
39
AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error)
40
}
41
42
-type GCBlockstore interface {
43
- Blockstore
44
-
42
+type GCLocker interface {
43
// GCLock locks the blockstore for garbage collection. No operations
44
// that expect to finish with a pin should ocurr simultaneously.
45
// Reading during GC is safe, and requires no lock.
@@ -58,6 +56,20 @@ type GCBlockstore interface {
56
GCRequested() bool
57
}
58
59
+type GCBlockstore interface {
60
+ Blockstore
61
+ GCLocker
62
+}
63
+
64
+func NewGCBlockstore(bs Blockstore, gcl GCLocker) GCBlockstore {
65
+ return gcBlockstore{bs, gcl}
66
+}
67
+
68
+type gcBlockstore struct {
69
+ Blockstore
70
+ GCLocker
71
+}
72
+
73
func NewBlockstore(d ds.Batching) *blockstore {
74
var dsb ds.Batching
75
dd := dsns.Wrap(d, BlockPrefix)
@@ -223,6 +235,16 @@ func (bs *blockstore) AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error)
235
return output, nil
236
}
237
238
+func NewGCLocker() *gclocker {
239
+ return &gclocker{}
240
+}
241
+
242
+type gclocker struct {
243
+ lk sync.RWMutex
244
+ gcreq int32
245
+ gcreqlk sync.Mutex
246
+}
247
+
248
type Unlocker interface {
249
Unlock()
250
}
@@ -236,18 +258,18 @@ func (u *unlocker) Unlock() {
258
u.unlock = nil // ensure its not called twice
259
}
260
239
-func (bs *blockstore) GCLock() Unlocker {
261
+func (bs *gclocker) GCLock() Unlocker {
262
atomic.AddInt32(&bs.gcreq, 1)
263
bs.lk.Lock()
264
atomic.AddInt32(&bs.gcreq, -1)
265
return &unlocker{bs.lk.Unlock}
266
}
267
246
-func (bs *blockstore) PinLock() Unlocker {
268
+func (bs *gclocker) PinLock() Unlocker {
269
bs.lk.RLock()
270
return &unlocker{bs.lk.RUnlock}
271
}
272
251
-func (bs *blockstore) GCRequested() bool {
273
+func (bs *gclocker) GCRequested() bool {
274
return atomic.LoadInt32(&bs.gcreq) > 0
275
}
blocks/blockstore/bloom_cache_test.go
+1
-1
@@ -14,7 +14,7 @@ import (
14
syncds "gx/ipfs/QmbzuUusHqaLLoNTDEVLcSF6vZDHZDLPC7p4bztRvvkXxU/go-datastore/sync"
15
)
16
17
-func testBloomCached(bs GCBlockstore, ctx context.Context) (*bloomcache, error) {
17
+func testBloomCached(bs Blockstore, ctx context.Context) (*bloomcache, error) {
18
if ctx == nil {
19
ctx = context.TODO()
20
}
blocks/blockstore/caching.go
+2
-2
@@ -22,8 +22,8 @@ func DefaultCacheOpts() CacheOpts {
22
}
23
}
24
25
-func CachedBlockstore(bs GCBlockstore,
26
- ctx context.Context, opts CacheOpts) (cbs GCBlockstore, err error) {
25
+func CachedBlockstore(bs Blockstore,
26
+ ctx context.Context, opts CacheOpts) (cbs Blockstore, err error) {
27
cbs = bs
28
29
if opts.HasBloomFilterSize < 0 || opts.HasBloomFilterHashes < 0 ||
blockservice/blockservice_test.go
+3
-3
@@ -36,14 +36,14 @@ func TestWriteThroughWorks(t *testing.T) {
36
}
37
}
38
39
-var _ blockstore.GCBlockstore = (*PutCountingBlockstore)(nil)
39
+var _ blockstore.Blockstore = (*PutCountingBlockstore)(nil)
40
41
type PutCountingBlockstore struct {
42
- blockstore.GCBlockstore
42
+ blockstore.Blockstore
43
PutCounter int
44
}
45
46
func (bs *PutCountingBlockstore) Put(block blocks.Block) error {
47
bs.PutCounter++
48
- return bs.GCBlockstore.Put(block)
48
+ return bs.Blockstore.Put(block)
49
}
core/builder.go
+3
-1
@@ -179,11 +179,13 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
179
opts.HasBloomFilterSize = 0
180
}
181
182
- n.Blockstore, err = bstore.CachedBlockstore(bs, ctx, opts)
182
+ cbs, err := bstore.CachedBlockstore(bs, ctx, opts)
183
if err != nil {
184
return err
185
}
186
187
+ n.Blockstore = bstore.NewGCBlockstore(cbs, bstore.NewGCLocker())
188
+
189
rcfg, err := n.Repo.Config()
190
if err != nil {
191
return err
unixfs/mod/dagmodifier_test.go
+1
-1
@@ -22,7 +22,7 @@ import (
22
"gx/ipfs/QmbzuUusHqaLLoNTDEVLcSF6vZDHZDLPC7p4bztRvvkXxU/go-datastore/sync"
23
)
24
25
-func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.GCBlockstore) {
25
+func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.Blockstore) {
26
dstore := ds.NewMapDatastore()
27
tsds := sync.MutexWrap(dstore)
28
bstore := blockstore.NewBlockstore(tsds)