cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

batch blockstore copies after block validation

Browse Source
This commit is contained in:
whyrusleeping 2020-09-23 18:53:28 -07:00
parent efd2dff0ca
commit b4e03d1759
1 changed files with 19 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
chain/sync.go
View File

@ -17,6 +17,7 @@ import (
"github.com/Gurpartap/async" "github.com/Gurpartap/async"
"github.com/hashicorp/go-multierror" "github.com/hashicorp/go-multierror"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid" "github.com/ipfs/go-cid"
cbor "github.com/ipfs/go-ipld-cbor" cbor "github.com/ipfs/go-ipld-cbor"
logging "github.com/ipfs/go-log/v2" logging "github.com/ipfs/go-log/v2"
@ -374,21 +375,28 @@ func (syncer *Syncer) InformNewBlock(from peer.ID, blk *types.FullBlock) bool {
return syncer.InformNewHead(from, fts) return syncer.InformNewHead(from, fts)
} }
func copyBlockstore(from, to bstore.Blockstore) error { func copyBlockstore(ctx context.Context, from, to bstore.Blockstore) error {
cids, err := from.AllKeysChan(context.TODO()) ctx, span := trace.StartSpan(ctx, "copyBlockstore")
defer span.End()
cids, err := from.AllKeysChan(ctx)
if err != nil { if err != nil {
return err return err
} }
// TODO: should probably expose better methods on the blockstore for this operation
var blks []blocks.Block
for c := range cids { for c := range cids {
b, err := from.Get(c) b, err := from.Get(c)
if err != nil { if err != nil {
return err return err
} }
if err := to.Put(b); err != nil { blks = append(blks, b)
return err }
}
if err := to.PutMany(blks); err != nil {
return err
} }
return nil return nil
@ -1515,11 +1523,11 @@ func (syncer *Syncer) iterFullTipsets(ctx context.Context, headers []*types.TipS
return err return err
} }
if err := persistMessages(bs, bstip); err != nil { if err := persistMessages(ctx, bs, bstip); err != nil {
return err return err
} }
if err := copyBlockstore(bs, syncer.store.Blockstore()); err != nil { if err := copyBlockstore(ctx, bs, syncer.store.Blockstore()); err != nil {
return xerrors.Errorf("message processing failed: %w", err) return xerrors.Errorf("message processing failed: %w", err)
} }
} }
@ -1596,7 +1604,10 @@ func (syncer *Syncer) fetchMessages(ctx context.Context, headers []*types.TipSet
return batch, nil return batch, nil
} }
func persistMessages(bs bstore.Blockstore, bst *exchange.CompactedMessages) error { func persistMessages(ctx context.Context, bs bstore.Blockstore, bst *exchange.CompactedMessages) error {
_, span := trace.StartSpan(ctx, "persistMessages")
defer span.End()
for _, m := range bst.Bls { for _, m := range bst.Bls {
//log.Infof("putting BLS message: %s", m.Cid()) //log.Infof("putting BLS message: %s", m.Cid())
if _, err := store.PutMessage(bs, m); err != nil { if _, err := store.PutMessage(bs, m); err != nil {