in-memory blockstore

Instead of using an in-memory datastore and dealing with the overhead of
computing datastore keys, creating new blocks, etc, use an in-memory blockstore.

lib/blockstore/blockstore.go

@@ -18,19 +18,18 @@ import (
 	"context"
 
 	ds "github.com/ipfs/go-datastore"
-	dssync "github.com/ipfs/go-datastore/sync"
 
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 )
 
 // NewTemporary returns a temporary blockstore.
-func NewTemporary() blockstore.Blockstore {
-	return NewBlockstore(ds.NewMapDatastore())
+func NewTemporary() MemStore {
+	return make(MemStore)
 }
 
 // NewTemporarySync returns a thread-safe temporary blockstore.
-func NewTemporarySync() blockstore.Blockstore {
-	return NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
+func NewTemporarySync() *SyncStore {
+	return &SyncStore{bs: make(MemStore)}
 }
 
 // WrapIDStore wraps the underlying blockstore in an "identity" blockstore.

lib/blockstore/memstore.go

@@ -0,0 +1,80 @@
+package blockstore
+
+import (
+	"context"
+
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	blockstore "github.com/ipfs/go-ipfs-blockstore"
+)
+
+type MemStore map[cid.Cid]blocks.Block
+
+func (m MemStore) DeleteBlock(k cid.Cid) error {
+	delete(m, k)
+	return nil
+}
+func (m MemStore) Has(k cid.Cid) (bool, error) {
+	_, ok := m[k]
+	return ok, nil
+}
+func (m MemStore) Get(k cid.Cid) (blocks.Block, error) {
+	b, ok := m[k]
+	if !ok {
+		return nil, blockstore.ErrNotFound
+	}
+	return b, nil
+}
+
+// GetSize returns the CIDs mapped BlockSize
+func (m MemStore) GetSize(k cid.Cid) (int, error) {
+	b, ok := m[k]
+	if !ok {
+		return 0, blockstore.ErrNotFound
+	}
+	return len(b.RawData()), nil
+}
+
+// Put puts a given block to the underlying datastore
+func (m MemStore) Put(b blocks.Block) error {
+	// Convert to a basic block for safety, but try to reuse the existing
+	// block if it's already a basic block.
+	k := b.Cid()
+	if _, ok := b.(*blocks.BasicBlock); !ok {
+		// If we already have the block, abort.
+		if _, ok := m[k]; ok {
+			return nil
+		}
+		// the error is only for debugging.
+		b, _ = blocks.NewBlockWithCid(b.RawData(), b.Cid())
+	}
+	m[b.Cid()] = b
+	return nil
+}
+
+// PutMany puts a slice of blocks at the same time using batching
+// capabilities of the underlying datastore whenever possible.
+func (m MemStore) PutMany(bs []blocks.Block) error {
+	for _, b := range bs {
+		_ = m.Put(b) // can't fail
+	}
+	return nil
+}
+
+// AllKeysChan returns a channel from which
+// the CIDs in the Blockstore can be read. It should respect
+// the given context, closing the channel if it becomes Done.
+func (m MemStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
+	ch := make(chan cid.Cid, len(m))
+	for k := range m {
+		ch <- k
+	}
+	close(ch)
+	return ch, nil
+}
+
+// HashOnRead specifies if every read block should be
+// rehashed to make sure it matches its CID.
+func (m MemStore) HashOnRead(enabled bool) {
+	// no-op
+}

lib/blockstore/syncstore.go

@@ -0,0 +1,68 @@
+package blockstore
+
+import (
+	"context"
+	"sync"
+
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+)
+
+type SyncStore struct {
+	mu sync.RWMutex
+	bs MemStore // specifically use a memStore to save indirection overhead.
+}
+
+func (m *SyncStore) DeleteBlock(k cid.Cid) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.bs.DeleteBlock(k)
+}
+func (m *SyncStore) Has(k cid.Cid) (bool, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.bs.Has(k)
+}
+func (m *SyncStore) Get(k cid.Cid) (blocks.Block, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.bs.Get(k)
+}
+
+// GetSize returns the CIDs mapped BlockSize
+func (m *SyncStore) GetSize(k cid.Cid) (int, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.bs.GetSize(k)
+}
+
+// Put puts a given block to the underlying datastore
+func (m *SyncStore) Put(b blocks.Block) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.bs.Put(b)
+}
+
+// PutMany puts a slice of blocks at the same time using batching
+// capabilities of the underlying datastore whenever possible.
+func (m *SyncStore) PutMany(bs []blocks.Block) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.bs.PutMany(bs)
+}
+
+// AllKeysChan returns a channel from which
+// the CIDs in the Blockstore can be read. It should respect
+// the given context, closing the channel if it becomes Done.
+func (m *SyncStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	// this blockstore implementation doesn't do any async work.
+	return m.bs.AllKeysChan(ctx)
+}
+
+// HashOnRead specifies if every read block should be
+// rehashed to make sure it matches its CID.
+func (m *SyncStore) HashOnRead(enabled bool) {
+	// noop
+}

lib/bufbstore/buf_bstore.go

@@ -19,10 +19,12 @@ type BufferedBS struct {
 }
 
 func NewBufferedBstore(base bstore.Blockstore) *BufferedBS {
-	buf := bstore.NewTemporary()
+	var buf bstore.Blockstore
 	if os.Getenv("LOTUS_DISABLE_VM_BUF") == "iknowitsabadidea" {
 		log.Warn("VM BLOCKSTORE BUFFERING IS DISABLED")
 		buf = base
+	} else {
+		buf = bstore.NewTemporary()
 	}
 
 	return &BufferedBS{