Merge pull request #10800 from filecoin-project/asr/speedup-persist

feat: chainstore: batch writes of tipsets

chain/store/snapshot.go

@@ -123,11 +123,9 @@ func (cs *ChainStore) Import(ctx context.Context, r io.Reader) (*types.TipSet, e
 	}
 
 	ts := root
+	tssToPersist := make([]*types.TipSet, 0, TipsetkeyBackfillRange)
 	for i := 0; i < int(TipsetkeyBackfillRange); i++ {
-		err = cs.PersistTipset(ctx, ts)
+		tssToPersist = append(tssToPersist, ts)
-		if err != nil {
-			return nil, err
-		}
 		parentTsKey := ts.Parents()
 		ts, err = cs.LoadTipSet(ctx, parentTsKey)
 		if ts == nil || err != nil {
@@ -136,6 +134,10 @@ func (cs *ChainStore) Import(ctx context.Context, r io.Reader) (*types.TipSet, e
 		}
 	}
 
+	if err := cs.PersistTipsets(ctx, tssToPersist); err != nil {
+		return nil, xerrors.Errorf("failed to persist tipsets: %w", err)
+	}
+
 	return root, nil
 }
 

chain/store/store.go

@@ -378,7 +378,7 @@ func (cs *ChainStore) SetGenesis(ctx context.Context, b *types.BlockHeader) erro
 }
 
 func (cs *ChainStore) PutTipSet(ctx context.Context, ts *types.TipSet) error {
-	if err := cs.PersistTipset(ctx, ts); err != nil {
+	if err := cs.PersistTipsets(ctx, []*types.TipSet{ts}); err != nil {
 		return xerrors.Errorf("failed to persist tipset: %w", err)
 	}
 
@@ -970,18 +970,25 @@ func (cs *ChainStore) AddToTipSetTracker(ctx context.Context, b *types.BlockHead
 	return nil
 }
 
-func (cs *ChainStore) PersistTipset(ctx context.Context, ts *types.TipSet) error {
+func (cs *ChainStore) PersistTipsets(ctx context.Context, tipsets []*types.TipSet) error {
-	if err := cs.persistBlockHeaders(ctx, ts.Blocks()...); err != nil {
+	toPersist := make([]*types.BlockHeader, 0, len(tipsets)*int(build.BlocksPerEpoch))
+	tsBlks := make([]block.Block, 0, len(tipsets))
+	for _, ts := range tipsets {
+		toPersist = append(toPersist, ts.Blocks()...)
+		tsBlk, err := ts.Key().ToStorageBlock()
+		if err != nil {
+			return xerrors.Errorf("failed to get tipset key block: %w", err)
+		}
+
+		tsBlks = append(tsBlks, tsBlk)
+	}
+
+	if err := cs.persistBlockHeaders(ctx, toPersist...); err != nil {
 		return xerrors.Errorf("failed to persist block headers: %w", err)
 	}
 
-	tsBlk, err := ts.Key().ToStorageBlock()
+	if err := cs.chainLocalBlockstore.PutMany(ctx, tsBlks); err != nil {
-	if err != nil {
+		return xerrors.Errorf("failed to put tipset key blocks: %w", err)
-		return xerrors.Errorf("failed to get tipset key block: %w", err)
-	}
-
-	if err = cs.chainLocalBlockstore.Put(ctx, tsBlk); err != nil {
-		return xerrors.Errorf("failed to put tipset key block: %w", err)
 	}
 
 	return nil

chain/sync.go

@@ -228,7 +228,7 @@ func (syncer *Syncer) InformNewHead(from peer.ID, fts *store.FullTipSet) bool {
 
 	// TODO: IMPORTANT(GARBAGE) this needs to be put in the 'temporary' side of
 	// the blockstore
-	if err := syncer.store.PersistTipset(ctx, fts.TipSet()); err != nil {
+	if err := syncer.store.PersistTipsets(ctx, []*types.TipSet{fts.TipSet()}); err != nil {
 		log.Warn("failed to persist incoming block header: ", err)
 		return false
 	}
@@ -1199,13 +1199,10 @@ func (syncer *Syncer) collectChain(ctx context.Context, ts *types.TipSet, hts *t
 	ss.SetStage(api.StagePersistHeaders)
 
 	// Write tipsets from oldest to newest.
-	for i := len(headers) - 1; i >= 0; i-- {
+	if err := syncer.store.PersistTipsets(ctx, headers); err != nil {
-		ts := headers[i]
+		err = xerrors.Errorf("failed to persist synced tipset to the chainstore: %w", err)
-		if err := syncer.store.PersistTipset(ctx, ts); err != nil {
+		ss.Error(err)
-			err = xerrors.Errorf("failed to persist synced tipset to the chainstore: %w", err)
+		return err
-			ss.Error(err)
-			return err
-		}
 	}
 
 	ss.SetStage(api.StageMessages)

cmd/lotus-sim/simulation/block.go

@@ -80,7 +80,7 @@ func (sim *Simulation) makeTipSet(ctx context.Context, messages []*types.Message
 		return nil, xerrors.Errorf("failed to create new tipset: %w", err)
 	}
 
-	err = sim.Node.Chainstore.PersistTipset(ctx, newTipSet)
+	err = sim.Node.Chainstore.PersistTipsets(ctx, []*types.TipSet{newTipSet})
 	if err != nil {
 		return nil, xerrors.Errorf("failed to persist block headers: %w", err)
 	}