lotus/blockstore/buffered.go

175 lines
3.4 KiB
Go
Raw Normal View History

package blockstore
2019-07-05 14:29:17 +00:00
import (
"context"
"os"
2019-07-05 14:29:17 +00:00
block "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
)
// buflog is a logger for the buffered blockstore. It is subscoped from the
// blockstore logger.
var buflog = log.Named("buf")
type BufferedBlockstore struct {
read Blockstore
write Blockstore
2019-07-05 14:29:17 +00:00
}
func NewBuffered(base Blockstore) *BufferedBlockstore {
var buf Blockstore
if os.Getenv("LOTUS_DISABLE_VM_BUF") == "iknowitsabadidea" {
buflog.Warn("VM BLOCKSTORE BUFFERING IS DISABLED")
buf = base
} else {
buf = NewMemory()
}
bs := &BufferedBlockstore{
2019-07-05 14:29:17 +00:00
read: base,
write: buf,
}
return bs
2019-07-05 14:29:17 +00:00
}
func NewTieredBstore(r Blockstore, w Blockstore) *BufferedBlockstore {
return &BufferedBlockstore{
read: r,
write: w,
}
}
2021-01-29 23:17:25 +00:00
var (
_ Blockstore = (*BufferedBlockstore)(nil)
_ Viewer = (*BufferedBlockstore)(nil)
)
2019-07-05 14:29:17 +00:00
func (bs *BufferedBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
2019-07-05 14:29:17 +00:00
a, err := bs.read.AllKeysChan(ctx)
if err != nil {
return nil, err
}
b, err := bs.write.AllKeysChan(ctx)
if err != nil {
return nil, err
}
out := make(chan cid.Cid)
go func() {
defer close(out)
for a != nil || b != nil {
select {
case val, ok := <-a:
if !ok {
a = nil
} else {
select {
case out <- val:
case <-ctx.Done():
return
}
}
case val, ok := <-b:
if !ok {
b = nil
} else {
select {
case out <- val:
case <-ctx.Done():
return
}
}
}
}
}()
return out, nil
}
func (bs *BufferedBlockstore) DeleteBlock(ctx context.Context, c cid.Cid) error {
if err := bs.read.DeleteBlock(ctx, c); err != nil {
2019-07-05 14:29:17 +00:00
return err
}
return bs.write.DeleteBlock(ctx, c)
2019-07-05 14:29:17 +00:00
}
func (bs *BufferedBlockstore) DeleteMany(ctx context.Context, cids []cid.Cid) error {
if err := bs.read.DeleteMany(ctx, cids); err != nil {
2021-03-02 14:45:45 +00:00
return err
}
return bs.write.DeleteMany(ctx, cids)
2021-03-02 14:45:45 +00:00
}
func (bs *BufferedBlockstore) View(ctx context.Context, c cid.Cid, callback func([]byte) error) error {
// both stores are viewable.
if err := bs.write.View(ctx, c, callback); err == ErrNotFound {
// not found in write blockstore; fall through.
} else {
return err // propagate errors, or nil, i.e. found.
}
return bs.read.View(ctx, c, callback)
}
func (bs *BufferedBlockstore) Get(ctx context.Context, c cid.Cid) (block.Block, error) {
if out, err := bs.write.Get(ctx, c); err != nil {
if err != ErrNotFound {
2019-07-05 14:29:17 +00:00
return nil, err
}
} else {
return out, nil
}
return bs.read.Get(ctx, c)
2019-07-05 14:29:17 +00:00
}
func (bs *BufferedBlockstore) GetSize(ctx context.Context, c cid.Cid) (int, error) {
s, err := bs.read.GetSize(ctx, c)
if err == ErrNotFound || s == 0 {
return bs.write.GetSize(ctx, c)
2020-07-22 19:26:57 +00:00
}
2020-07-23 08:23:44 +00:00
return s, err
2019-07-05 14:29:17 +00:00
}
func (bs *BufferedBlockstore) Put(ctx context.Context, blk block.Block) error {
has, err := bs.read.Has(ctx, blk.Cid()) // TODO: consider dropping this check
if err != nil {
return err
}
if has {
return nil
}
return bs.write.Put(ctx, blk)
2019-07-05 14:29:17 +00:00
}
func (bs *BufferedBlockstore) Has(ctx context.Context, c cid.Cid) (bool, error) {
has, err := bs.write.Has(ctx, c)
2019-07-05 14:29:17 +00:00
if err != nil {
return false, err
}
if has {
return true, nil
}
return bs.read.Has(ctx, c)
2019-07-05 14:29:17 +00:00
}
2021-12-11 21:03:00 +00:00
func (bs *BufferedBlockstore) HashOnRead(hor bool) {
bs.read.HashOnRead(hor)
bs.write.HashOnRead(hor)
2019-07-05 14:29:17 +00:00
}
func (bs *BufferedBlockstore) PutMany(ctx context.Context, blks []block.Block) error {
return bs.write.PutMany(ctx, blks)
2019-07-05 14:36:08 +00:00
}
func (bs *BufferedBlockstore) Read() Blockstore {
return bs.read
}