Merge pull request #466 from filecoin-project/feat/async-validateblk

sync: Do some things in parallel in ValidateBlock

chain/sync.go

@@ -7,18 +7,10 @@ import (
 	"sync"
 	"time"
 
-	"github.com/filecoin-project/go-bls-sigs"
+	"github.com/Gurpartap/async"
-	"github.com/filecoin-project/lotus/api"
-	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain/actors"
-	"github.com/filecoin-project/lotus/chain/address"
-	"github.com/filecoin-project/lotus/chain/blocksync"
-	"github.com/filecoin-project/lotus/chain/state"
-	"github.com/filecoin-project/lotus/chain/stmgr"
-	"github.com/filecoin-project/lotus/chain/store"
-	"github.com/filecoin-project/lotus/chain/types"
-
 	amt "github.com/filecoin-project/go-amt-ipld"
+	"github.com/filecoin-project/go-bls-sigs"
+	"github.com/hashicorp/go-multierror"
 	"github.com/ipfs/go-cid"
 	dstore "github.com/ipfs/go-datastore"
 	hamt "github.com/ipfs/go-hamt-ipld"
@@ -28,6 +20,16 @@ import (
 	cbg "github.com/whyrusleeping/cbor-gen"
 	"go.opencensus.io/trace"
 	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/address"
+	"github.com/filecoin-project/lotus/chain/blocksync"
+	"github.com/filecoin-project/lotus/chain/state"
+	"github.com/filecoin-project/lotus/chain/stmgr"
+	"github.com/filecoin-project/lotus/chain/store"
+	"github.com/filecoin-project/lotus/chain/types"
 )
 
 var log = logging.Logger("chain")
@@ -482,6 +484,43 @@ func (syncer *Syncer) ValidateBlock(ctx context.Context, b *types.FullBlock) err
 		return xerrors.Errorf("load parent tipset failed (%s): %w", h.Parents, err)
 	}
 
+	// fast checks first
+	if h.Timestamp > uint64(time.Now().Unix()+build.AllowableClockDrift) {
+		return xerrors.Errorf("block was from the future")
+	}
+
+	if h.Timestamp < baseTs.MinTimestamp()+uint64(build.BlockDelay*len(h.Tickets)) {
+		log.Warn("timestamp funtimes: ", h.Timestamp, baseTs.MinTimestamp(), len(h.Tickets))
+		return xerrors.Errorf("block was generated too soon (h.ts:%d < base.mints:%d + BLOCK_DELAY:%d * tkts.len:%d)", h.Timestamp, baseTs.MinTimestamp(), build.BlockDelay, len(h.Tickets))
+	}
+
+	winnerCheck := async.Err(func() error {
+		mpow, tpow, err := stmgr.GetPower(ctx, syncer.sm, baseTs, h.Miner)
+		if err != nil {
+			return xerrors.Errorf("failed getting power: %w", err)
+		}
+
+		if !types.PowerCmp(h.ElectionProof, mpow, tpow) {
+			return xerrors.Errorf("miner created a block but was not a winner")
+		}
+		return nil
+	})
+
+	msgsCheck := async.Err(func() error {
+		if err := syncer.checkBlockMessages(ctx, b, baseTs); err != nil {
+			return xerrors.Errorf("block had invalid messages: %w", err)
+		}
+		return nil
+	})
+
+	minerCheck := async.Err(func() error {
+		if err := syncer.minerIsValid(ctx, h.Miner, baseTs); err != nil {
+			return xerrors.Errorf("minerIsValid failed: %w", err)
+		}
+		return nil
+	})
+
+	// Stuff that needs stateroot / worker address
 	stateroot, precp, err := syncer.sm.TipSetState(ctx, baseTs)
 	if err != nil {
 		return xerrors.Errorf("get tipsetstate(%d, %s) failed: %w", h.Height, h.Parents, err)
@@ -506,32 +545,26 @@ func (syncer *Syncer) ValidateBlock(ctx context.Context, b *types.FullBlock) err
 		return xerrors.Errorf("parent receipts root did not match computed value (%s != %s)", precp, h.ParentMessageReceipts)
 	}
 
-	if h.Timestamp > uint64(time.Now().Unix()+build.AllowableClockDrift) {
-		return xerrors.Errorf("block was from the future: %w", ErrTemporal)
-	}
-
-	if h.Timestamp < baseTs.MinTimestamp()+uint64(build.BlockDelay*len(h.Tickets)) {
-		log.Warn("timestamp funtimes: ", h.Timestamp, baseTs.MinTimestamp(), len(h.Tickets))
-		return xerrors.Errorf("block was generated too soon (h.ts:%d < base.mints:%d + BLOCK_DELAY:%d * tkts.len:%d)", h.Timestamp, baseTs.MinTimestamp(), build.BlockDelay, len(h.Tickets))
-	}
-
-	if err := syncer.minerIsValid(ctx, h.Miner, baseTs); err != nil {
-		return xerrors.Errorf("minerIsValid failed: %w", err)
-	}
-
 	waddr, err := stmgr.GetMinerWorkerRaw(ctx, syncer.sm, stateroot, h.Miner)
 	if err != nil {
 		return xerrors.Errorf("GetMinerWorkerRaw failed: %w", err)
 	}
 
+	blockSigCheck := async.Err(func() error {
 		if err := h.CheckBlockSignature(ctx, waddr); err != nil {
 			return xerrors.Errorf("check block signature failed: %w", err)
 		}
+		return nil
+	})
 
+	tktsCheck := async.Err(func() error {
 		if err := syncer.validateTickets(ctx, waddr, h.Tickets, baseTs); err != nil {
 			return xerrors.Errorf("validating block tickets failed: %w", err)
 		}
+		return nil
+	})
 
+	eproofCheck := async.Err(func() error {
 		rand, err := syncer.sm.ChainStore().GetRandomness(ctx, baseTs.Cids(), h.Tickets, build.EcRandomnessLookback)
 		if err != nil {
 			return xerrors.Errorf("failed to get randomness for verifying election proof: %w", err)
@@ -540,21 +573,26 @@ func (syncer *Syncer) ValidateBlock(ctx context.Context, b *types.FullBlock) err
 		if err := VerifyElectionProof(ctx, h.ElectionProof, rand, waddr); err != nil {
 			return xerrors.Errorf("checking eproof failed: %w", err)
 		}
-
-	mpow, tpow, err := stmgr.GetPower(ctx, syncer.sm, baseTs, h.Miner)
-	if err != nil {
-		return xerrors.Errorf("failed getting power: %w", err)
-	}
-
-	if !types.PowerCmp(h.ElectionProof, mpow, tpow) {
-		return xerrors.Errorf("miner created a block but was not a winner")
-	}
-
-	if err := syncer.checkBlockMessages(ctx, b, baseTs); err != nil {
-		return xerrors.Errorf("block had invalid messages: %w", err)
-	}
-
 		return nil
+	})
+
+	await := []async.ErrorFuture{
+		minerCheck,
+		tktsCheck,
+		blockSigCheck,
+		eproofCheck,
+		winnerCheck,
+		msgsCheck,
+	}
+
+	var merr error
+	for _, fut := range await {
+		if err := fut.AwaitContext(ctx); err != nil {
+			err = multierror.Append(merr, err)
+		}
+	}
+
+	return merr
 }
 
 func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock, baseTs *types.TipSet) error {
@@ -833,6 +871,7 @@ func (syncer *Syncer) syncFork(ctx context.Context, from *types.TipSet, to *type
 
 func (syncer *Syncer) syncMessagesAndCheckState(ctx context.Context, headers []*types.TipSet) error {
 	syncer.syncState.SetHeight(0)
+
 	return syncer.iterFullTipsets(ctx, headers, func(ctx context.Context, fts *store.FullTipSet) error {
 		log.Debugw("validating tipset", "height", fts.TipSet().Height(), "size", len(fts.TipSet().Cids()))
 		if err := syncer.ValidateTipSet(ctx, fts); err != nil {

go.mod

@@ -6,6 +6,7 @@ require (
 	contrib.go.opencensus.io/exporter/jaeger v0.1.0
 	github.com/BurntSushi/toml v0.3.1
 	github.com/GeertJohan/go.rice v1.0.0
+	github.com/Gurpartap/async v0.0.0-20180927173644-4f7f499dd9ee
 	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
 	github.com/fatih/color v1.7.0 // indirect
 	github.com/filecoin-project/go-amt-ipld v0.0.0-20190919045431-3650716fff16
@@ -15,6 +16,7 @@ require (
 	github.com/gbrlsnchs/jwt/v3 v3.0.0-beta.1
 	github.com/go-ole/go-ole v1.2.4 // indirect
 	github.com/gorilla/websocket v1.4.0
+	github.com/hashicorp/go-multierror v1.0.0
 	github.com/hashicorp/golang-lru v0.5.3
 	github.com/influxdata/influxdb1-client v0.0.0-20190809212627-fc22c7df067e
 	github.com/ipfs/go-bitswap v0.1.8

go.sum

@@ -12,6 +12,8 @@ github.com/GeertJohan/go.incremental v1.0.0 h1:7AH+pY1XUgQE4Y1HcXYaMqAI0m9yrFqo/
 github.com/GeertJohan/go.incremental v1.0.0/go.mod h1:6fAjUhbVuX1KcMD3c8TEgVUqmo4seqhv0i0kdATSkM0=
 github.com/GeertJohan/go.rice v1.0.0 h1:KkI6O9uMaQU3VEKaj01ulavtF7o1fWT7+pk/4voiMLQ=
 github.com/GeertJohan/go.rice v1.0.0/go.mod h1:eH6gbSOAUv07dQuZVnBmoDP8mgsM1rtixis4Tib9if0=
+github.com/Gurpartap/async v0.0.0-20180927173644-4f7f499dd9ee h1:8doiS7ib3zi6/K172oDhSKU0dJ/miJramo9NITOMyZQ=
+github.com/Gurpartap/async v0.0.0-20180927173644-4f7f499dd9ee/go.mod h1:W0GbEAA4uFNYOGG2cJpmFJ04E6SD1NLELPYZB57/7AY=
 github.com/Kubuxu/go-os-helper v0.0.1/go.mod h1:N8B+I7vPCT80IcP58r50u4+gEEcsZETFUpAzWW2ep1Y=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=