From 8f8de2d42ec835414ef3e2d94afc5c217d3a123b Mon Sep 17 00:00:00 2001 From: Jakub Sztandera Date: Mon, 4 Jul 2016 23:02:29 +0200 Subject: [PATCH] blocks/blockstore: introduce context passing to blockstore License: MIT Signed-off-by: Jakub Sztandera --- blocks/blockstore/bloom_cache.go | 26 ++++++++++++++++++-------- blocks/blockstore/bloom_cache_test.go | 25 +++++++++++++++++++------ blocks/blockstore/caching.go | 26 ++++++++++++++++++++++++++ core/builder.go | 2 +- exchange/bitswap/testutils.go | 3 ++- 5 files changed, 66 insertions(+), 16 deletions(-) diff --git a/blocks/blockstore/bloom_cache.go b/blocks/blockstore/bloom_cache.go index 3e4038869a97..d9f93b3e8550 100644 --- a/blocks/blockstore/bloom_cache.go +++ b/blocks/blockstore/bloom_cache.go @@ -11,10 +11,10 @@ import ( "sync/atomic" ) -// BloomCached returns Blockstore that caches Has requests using Bloom filter +// bloomCached returns Blockstore that caches Has requests using Bloom filter // Size is size of bloom filter in bytes -func BloomCached(bs Blockstore, bloomSize, lruSize int) (*bloomcache, error) { - bl, err := bloom.New(float64(bloomSize), float64(7)) +func bloomCached(bs Blockstore, ctx context.Context, bloomSize, hashCount, lruSize int) (*bloomcache, error) { + bl, err := bloom.New(float64(bloomSize), float64(hashCount)) if err != nil { return nil, err } @@ -24,7 +24,7 @@ func BloomCached(bs Blockstore, bloomSize, lruSize int) (*bloomcache, error) { } bc := &bloomcache{blockstore: bs, bloom: bl, arc: arc} bc.Invalidate() - go bc.Rebuild() + go bc.Rebuild(ctx) return bc, nil } @@ -52,8 +52,7 @@ func (b *bloomcache) BloomActive() bool { return atomic.LoadInt32(&b.active) != 0 } -func (b *bloomcache) Rebuild() { - ctx := context.TODO() +func (b *bloomcache) Rebuild(ctx context.Context) { evt := log.EventBegin(ctx, "bloomcache.Rebuild") defer evt.Done() @@ -62,8 +61,19 @@ func (b *bloomcache) Rebuild() { log.Errorf("AllKeysChan failed in bloomcache rebuild with: %v", err) return } - for key := range ch { - b.bloom.AddTS([]byte(key)) // Use binary key, the more compact the better + finish := false + for !finish { + select { + case key, ok := <-ch: + if ok { + b.bloom.AddTS([]byte(key)) // Use binary key, the more compact the better + } else { + finish = true + } + case <-ctx.Done(): + log.Warning("Cache rebuild closed by context finishing.") + return + } } close(b.rebuildChan) atomic.StoreInt32(&b.active, 1) diff --git a/blocks/blockstore/bloom_cache_test.go b/blocks/blockstore/bloom_cache_test.go index 3b8bb8b91161..c22b373bda2e 100644 --- a/blocks/blockstore/bloom_cache_test.go +++ b/blocks/blockstore/bloom_cache_test.go @@ -8,18 +8,30 @@ import ( "github.com/ipfs/go-ipfs/blocks" + context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context" ds "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore" dsq "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore/query" syncds "gx/ipfs/QmfQzVugPq1w5shWRcLWSeiHF4a2meBX7yVD8Vw7GWJM9o/go-datastore/sync" ) +func testBloomCached(bs GCBlockstore, ctx context.Context) (*bloomcache, error) { + opts := DefaultCacheOpts() + opts.BlockARCCacheSize = 0 + bbs, err := CachedBlockstore(bs, ctx, opts) + if err == nil { + return bbs.(*bloomcache), nil + } else { + return nil, err + } +} + func TestReturnsErrorWhenSizeNegative(t *testing.T) { bs := NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore())) - _, err := BloomCached(bs, 100, -1) + _, err := bloomCached(bs, nil, 100, 1, -1) if err == nil { t.Fail() } - _, err = BloomCached(bs, -1, 100) + _, err = bloomCached(bs, nil, -1, 1, 100) if err == nil { t.Fail() } @@ -29,7 +41,7 @@ func TestRemoveCacheEntryOnDelete(t *testing.T) { b := blocks.NewBlock([]byte("foo")) cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()} bs := NewBlockstore(syncds.MutexWrap(cd)) - cachedbs, err := BloomCached(bs, 1, 1) + cachedbs, err := testBloomCached(bs, nil) if err != nil { t.Fatal(err) } @@ -53,7 +65,7 @@ func TestRemoveCacheEntryOnDelete(t *testing.T) { func TestElideDuplicateWrite(t *testing.T) { cd := &callbackDatastore{f: func() {}, ds: ds.NewMapDatastore()} bs := NewBlockstore(syncds.MutexWrap(cd)) - cachedbs, err := BloomCached(bs, 1, 1) + cachedbs, err := testBloomCached(bs, nil) if err != nil { t.Fatal(err) } @@ -73,14 +85,15 @@ func TestHasIsBloomCached(t *testing.T) { for i := 0; i < 1000; i++ { bs.Put(blocks.NewBlock([]byte(fmt.Sprintf("data: %d", i)))) } - cachedbs, err := BloomCached(bs, 256*1024, 128) + ctx, _ := context.WithTimeout(context.Background(), 1*time.Second) + cachedbs, err := testBloomCached(bs, ctx) if err != nil { t.Fatal(err) } select { case <-cachedbs.rebuildChan: - case <-time.After(1 * time.Second): + case <-ctx.Done(): t.Fatalf("Timeout wating for rebuild: %d", cachedbs.bloom.ElementsAdded()) } diff --git a/blocks/blockstore/caching.go b/blocks/blockstore/caching.go index 5657059d89b8..9a7bf9b23413 100644 --- a/blocks/blockstore/caching.go +++ b/blocks/blockstore/caching.go @@ -1,5 +1,11 @@ package blockstore +import ( + "errors" + + context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context" +) + // Next to each option is it aproximate memory usage per unit type CacheOpts struct { HasBloomFilterSize int // 1 bit @@ -14,3 +20,23 @@ func DefaultCacheOpts() CacheOpts { HasARCCacheSize: 64 * 1024, } } + +func CachedBlockstore(bs GCBlockstore, + ctx context.Context, opts CacheOpts) (cbs GCBlockstore, err error) { + if ctx == nil { + ctx = context.TODO() // For tests + } + + if opts.HasBloomFilterSize < 0 || opts.HasBloomFilterHashes < 0 || + opts.HasARCCacheSize < 0 || opts.BlockARCCacheSize < 0 { + return nil, errors.New("all options for cache need to be greater than zero") + } + + if opts.HasBloomFilterSize != 0 && opts.HasBloomFilterHashes == 0 { + return nil, errors.New("bloom filter hash count can't be 0 when there is size set") + } + cbs, err = bloomCached(bs, ctx, opts.HasBloomFilterSize, opts.HasBloomFilterHashes, + opts.HasARCCacheSize) + + return cbs, err +} diff --git a/core/builder.go b/core/builder.go index f602db1d6b14..6095e2c24879 100644 --- a/core/builder.go +++ b/core/builder.go @@ -131,7 +131,7 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error { var err error bs := bstore.NewBlockstore(n.Repo.Datastore()) - n.Blockstore, err = bstore.BloomCached(bs, 256*1024, kSizeBlockstoreWriteCache) + n.Blockstore, err = bstore.CachedBlockstore(bs, ctx, bstore.DefaultCacheOpts()) if err != nil { return err } diff --git a/exchange/bitswap/testutils.go b/exchange/bitswap/testutils.go index d42278e710cc..a4be8d06fa31 100644 --- a/exchange/bitswap/testutils.go +++ b/exchange/bitswap/testutils.go @@ -93,7 +93,8 @@ func Session(ctx context.Context, net tn.Network, p testutil.Identity) Instance adapter := net.Adapter(p) dstore := ds_sync.MutexWrap(datastore2.WithDelay(ds.NewMapDatastore(), bsdelay)) - bstore, err := blockstore.BloomCached(blockstore.NewBlockstore(ds_sync.MutexWrap(dstore)), bloomSize, writeCacheElems) + bstore, err := blockstore.CachedBlockstore(blockstore.NewBlockstore( + ds_sync.MutexWrap(dstore)), ctx, blockstore.DefaultCacheOpts()) if err != nil { panic(err.Error()) // FIXME perhaps change signature and return error. }