From ffe9d7dae447e239c5d61478b9e0cd3ef7cf216c Mon Sep 17 00:00:00 2001 From: Kevin Atkinson Date: Wed, 2 Nov 2016 21:56:34 -0400 Subject: [PATCH] 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 --- blocks/blockstore/arc_cache_test.go | 2 +- blocks/blockstore/blockstore.go | 34 ++++++++++++++++++++++----- blocks/blockstore/bloom_cache_test.go | 2 +- blocks/blockstore/caching.go | 4 ++-- blockservice/blockservice_test.go | 6 ++--- core/builder.go | 4 +++- unixfs/mod/dagmodifier_test.go | 2 +- 7 files changed, 39 insertions(+), 15 deletions(-) diff --git a/blocks/blockstore/arc_cache_test.go b/blocks/blockstore/arc_cache_test.go index d796214c5ca..0f7823c5cc8 100644 --- a/blocks/blockstore/arc_cache_test.go +++ b/blocks/blockstore/arc_cache_test.go @@ -13,7 +13,7 @@ import ( var exampleBlock = blocks.NewBlock([]byte("foo")) -func testArcCached(bs GCBlockstore, ctx context.Context) (*arccache, error) { +func testArcCached(bs Blockstore, ctx context.Context) (*arccache, error) { if ctx == nil { ctx = context.TODO() } diff --git a/blocks/blockstore/blockstore.go b/blocks/blockstore/blockstore.go index f5dc26c1c4e..274c1ee7b4f 100644 --- a/blocks/blockstore/blockstore.go +++ b/blocks/blockstore/blockstore.go @@ -39,9 +39,7 @@ type Blockstore interface { AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) } -type GCBlockstore interface { - Blockstore - +type GCLocker interface { // GCLock locks the blockstore for garbage collection. No operations // that expect to finish with a pin should ocurr simultaneously. // Reading during GC is safe, and requires no lock. @@ -58,6 +56,20 @@ type GCBlockstore interface { GCRequested() bool } +type GCBlockstore interface { + Blockstore + GCLocker +} + +func NewGCBlockstore(bs Blockstore, gcl GCLocker) GCBlockstore { + return gcBlockstore{bs, gcl} +} + +type gcBlockstore struct { + Blockstore + GCLocker +} + func NewBlockstore(d ds.Batching) *blockstore { var dsb ds.Batching dd := dsns.Wrap(d, BlockPrefix) @@ -223,6 +235,16 @@ func (bs *blockstore) AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) return output, nil } +func NewGCLocker() *gclocker { + return &gclocker{} +} + +type gclocker struct { + lk sync.RWMutex + gcreq int32 + gcreqlk sync.Mutex +} + type Unlocker interface { Unlock() } @@ -236,18 +258,18 @@ func (u *unlocker) Unlock() { u.unlock = nil // ensure its not called twice } -func (bs *blockstore) GCLock() Unlocker { +func (bs *gclocker) GCLock() Unlocker { atomic.AddInt32(&bs.gcreq, 1) bs.lk.Lock() atomic.AddInt32(&bs.gcreq, -1) return &unlocker{bs.lk.Unlock} } -func (bs *blockstore) PinLock() Unlocker { +func (bs *gclocker) PinLock() Unlocker { bs.lk.RLock() return &unlocker{bs.lk.RUnlock} } -func (bs *blockstore) GCRequested() bool { +func (bs *gclocker) GCRequested() bool { return atomic.LoadInt32(&bs.gcreq) > 0 } diff --git a/blocks/blockstore/bloom_cache_test.go b/blocks/blockstore/bloom_cache_test.go index 8bdf567f07a..72223cd44e0 100644 --- a/blocks/blockstore/bloom_cache_test.go +++ b/blocks/blockstore/bloom_cache_test.go @@ -14,7 +14,7 @@ import ( syncds "gx/ipfs/QmbzuUusHqaLLoNTDEVLcSF6vZDHZDLPC7p4bztRvvkXxU/go-datastore/sync" ) -func testBloomCached(bs GCBlockstore, ctx context.Context) (*bloomcache, error) { +func testBloomCached(bs Blockstore, ctx context.Context) (*bloomcache, error) { if ctx == nil { ctx = context.TODO() } diff --git a/blocks/blockstore/caching.go b/blocks/blockstore/caching.go index d28401cf8a4..d19f4782267 100644 --- a/blocks/blockstore/caching.go +++ b/blocks/blockstore/caching.go @@ -22,8 +22,8 @@ func DefaultCacheOpts() CacheOpts { } } -func CachedBlockstore(bs GCBlockstore, - ctx context.Context, opts CacheOpts) (cbs GCBlockstore, err error) { +func CachedBlockstore(bs Blockstore, + ctx context.Context, opts CacheOpts) (cbs Blockstore, err error) { cbs = bs if opts.HasBloomFilterSize < 0 || opts.HasBloomFilterHashes < 0 || diff --git a/blockservice/blockservice_test.go b/blockservice/blockservice_test.go index d87a383e566..0415f8213d2 100644 --- a/blockservice/blockservice_test.go +++ b/blockservice/blockservice_test.go @@ -36,14 +36,14 @@ func TestWriteThroughWorks(t *testing.T) { } } -var _ blockstore.GCBlockstore = (*PutCountingBlockstore)(nil) +var _ blockstore.Blockstore = (*PutCountingBlockstore)(nil) type PutCountingBlockstore struct { - blockstore.GCBlockstore + blockstore.Blockstore PutCounter int } func (bs *PutCountingBlockstore) Put(block blocks.Block) error { bs.PutCounter++ - return bs.GCBlockstore.Put(block) + return bs.Blockstore.Put(block) } diff --git a/core/builder.go b/core/builder.go index 451e771d32c..baef82ed06d 100644 --- a/core/builder.go +++ b/core/builder.go @@ -179,11 +179,13 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error { opts.HasBloomFilterSize = 0 } - n.Blockstore, err = bstore.CachedBlockstore(bs, ctx, opts) + cbs, err := bstore.CachedBlockstore(bs, ctx, opts) if err != nil { return err } + n.Blockstore = bstore.NewGCBlockstore(cbs, bstore.NewGCLocker()) + rcfg, err := n.Repo.Config() if err != nil { return err diff --git a/unixfs/mod/dagmodifier_test.go b/unixfs/mod/dagmodifier_test.go index 810ec6f2344..9c7ac89d7be 100644 --- a/unixfs/mod/dagmodifier_test.go +++ b/unixfs/mod/dagmodifier_test.go @@ -22,7 +22,7 @@ import ( "gx/ipfs/QmbzuUusHqaLLoNTDEVLcSF6vZDHZDLPC7p4bztRvvkXxU/go-datastore/sync" ) -func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.GCBlockstore) { +func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.Blockstore) { dstore := ds.NewMapDatastore() tsds := sync.MutexWrap(dstore) bstore := blockstore.NewBlockstore(tsds)