Parallelize BatchedParseBlock (#3227)

Signed-off-by: Yacov Manevich <yacov.manevich@avalabs.org>

Author: yacovm

vms/proposervm/batched_vm.go

@@ -81,10 +81,6 @@ func (vm *VM) GetAncestors(
 }
 
 func (vm *VM) BatchedParseBlock(ctx context.Context, blks [][]byte) ([]snowman.Block, error) {
-	if vm.batchedVM == nil {
-		return nil, block.ErrRemoteVMNotImplemented
-	}
-
 	type partialData struct {
 		index int
 		block statelessblock.Block
@@ -97,9 +93,11 @@ func (vm *VM) BatchedParseBlock(ctx context.Context, blks [][]byte) ([]snowman.B
 		statelessBlockDescs = make([]partialData, 0, len(blks))
 		innerBlockBytes     = make([][]byte, 0, len(blks))
 	)
+
+	parsingResults := statelessblock.ParseBlocks(blks, vm.ctx.ChainID)
+
 	for ; blocksIndex < len(blks); blocksIndex++ {
-		blkBytes := blks[blocksIndex]
-		statelessBlock, err := statelessblock.Parse(blkBytes, vm.ctx.ChainID)
+		statelessBlock, err := parsingResults[blocksIndex].Block, parsingResults[blocksIndex].Err
 		if err != nil {
 			break
 		}
@@ -120,7 +118,7 @@ func (vm *VM) BatchedParseBlock(ctx context.Context, blks [][]byte) ([]snowman.B
 	innerBlockBytes = append(innerBlockBytes, blks[blocksIndex:]...)
 
 	// parse all inner blocks at once
-	innerBlks, err := vm.batchedVM.BatchedParseBlock(ctx, innerBlockBytes)
+	innerBlks, err := block.BatchedParseBlock(ctx, vm.ChainVM, innerBlockBytes)
 	if err != nil {
 		return nil, err
 	}

vms/proposervm/batched_vm_test.go

@@ -6,6 +6,8 @@ package proposervm
 import (
 	"bytes"
 	"context"
+	"crypto"
+	"encoding/binary"
 	"testing"
 	"time"
 
@@ -23,7 +25,10 @@ import (
 	"github.com/ava-labs/avalanchego/snow/engine/snowman/block"
 	"github.com/ava-labs/avalanchego/snow/snowtest"
 	"github.com/ava-labs/avalanchego/snow/validators"
+	"github.com/ava-labs/avalanchego/staking"
 	"github.com/ava-labs/avalanchego/utils/timer/mockable"
+
+	blockbuilder "github.com/ava-labs/avalanchego/vms/proposervm/block"
 )
 
 func TestCoreVMNotRemote(t *testing.T) {
@@ -52,8 +57,9 @@ func TestCoreVMNotRemote(t *testing.T) {
 	require.ErrorIs(err, block.ErrRemoteVMNotImplemented)
 
 	var blks [][]byte
-	_, err = proVM.BatchedParseBlock(context.Background(), blks)
-	require.ErrorIs(err, block.ErrRemoteVMNotImplemented)
+	shouldBeEmpty, err := proVM.BatchedParseBlock(context.Background(), blks)
+	require.NoError(err)
+	require.Empty(shouldBeEmpty)
 }
 
 func TestGetAncestorsPreForkOnly(t *testing.T) {
@@ -583,6 +589,102 @@ func TestBatchedParseBlockPreForkOnly(t *testing.T) {
 	require.Equal(builtBlk3.ID(), res[2].ID())
 }
 
+func TestBatchedParseBlockParallel(t *testing.T) {
+	parentID := ids.ID{1}
+	timestamp := time.Unix(123, 0)
+	pChainHeight := uint64(2)
+	chainID := ids.GenerateTestID()
+
+	vm := VM{
+		ctx: &snow.Context{ChainID: chainID},
+		ChainVM: &block.TestVM{
+			ParseBlockF: func(_ context.Context, rawBlock []byte) (snowman.Block, error) {
+				return &snowmantest.Block{BytesV: rawBlock}, nil
+			},
+		},
+	}
+
+	tlsCert, err := staking.NewTLSCert()
+	require.NoError(t, err)
+
+	cert, err := staking.ParseCertificate(tlsCert.Leaf.Raw)
+	require.NoError(t, err)
+	key := tlsCert.PrivateKey.(crypto.Signer)
+
+	blockThatCantBeParsed := snowmantest.BuildChild(snowmantest.Genesis)
+
+	blocksWithUnparsable := makeParseableBlocks(t, parentID, timestamp, pChainHeight, cert, chainID, key)
+	blocksWithUnparsable[50] = blockThatCantBeParsed.Bytes()
+
+	parsableBlocks := makeParseableBlocks(t, parentID, timestamp, pChainHeight, cert, chainID, key)
+
+	for _, testCase := range []struct {
+		name         string
+		preForkIndex int
+		rawBlocks    [][]byte
+	}{
+		{
+			name:      "empty input",
+			rawBlocks: [][]byte{},
+		},
+		{
+			name:         "pre-fork is somewhere in the middle",
+			rawBlocks:    blocksWithUnparsable,
+			preForkIndex: 50,
+		},
+		{
+			name:         "all blocks are post fork",
+			rawBlocks:    parsableBlocks,
+			preForkIndex: len(parsableBlocks),
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			require := require.New(t)
+			blocks, err := vm.BatchedParseBlock(context.Background(), testCase.rawBlocks)
+			require.NoError(err)
+
+			returnedBlockBytes := make([][]byte, len(blocks))
+			for i, block := range blocks {
+				returnedBlockBytes[i] = block.Bytes()
+			}
+			require.Equal(testCase.rawBlocks, returnedBlockBytes)
+
+			for i, block := range blocks {
+				if i < testCase.preForkIndex {
+					require.IsType(&postForkBlock{}, block)
+				} else {
+					require.IsType(&preForkBlock{}, block)
+				}
+			}
+		})
+	}
+}
+
+func makeParseableBlocks(t *testing.T, parentID ids.ID, timestamp time.Time, pChainHeight uint64, cert *staking.Certificate, chainID ids.ID, key crypto.Signer) [][]byte {
+	makeSignedBlock := func(i int) []byte {
+		buff := binary.AppendVarint(nil, int64(i))
+
+		signedBlock, err := blockbuilder.Build(
+			parentID,
+			timestamp,
+			pChainHeight,
+			cert,
+			buff,
+			chainID,
+			key,
+		)
+		require.NoError(t, err)
+
+		return signedBlock.Bytes()
+	}
+
+	blockBytes := make([][]byte, 100)
+	for i := range blockBytes {
+		blockBytes[i] = makeSignedBlock(i)
+	}
+	return blockBytes
+}
+
 func TestBatchedParseBlockPostForkOnly(t *testing.T) {
 	require := require.New(t)
 	var (

vms/proposervm/block/parse.go

@@ -5,10 +5,36 @@ package block
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/ava-labs/avalanchego/ids"
 )
 
+type ParseResult struct {
+	Block Block
+	Err   error
+}
+
+// ParseBlocks parses the given raw blocks into tuples of (Block, error).
+// Each ParseResult is returned in the same order as its corresponding bytes in the input.
+func ParseBlocks(blks [][]byte, chainID ids.ID) []ParseResult {
+	results := make([]ParseResult, len(blks))
+
+	var wg sync.WaitGroup
+	wg.Add(len(blks))
+
+	for i, blk := range blks {
+		go func(i int, blkBytes []byte) {
+			defer wg.Done()
+			results[i].Block, results[i].Err = Parse(blkBytes, chainID)
+		}(i, blk)
+	}
+
+	wg.Wait()
+
+	return results
+}
+
 // Parse a block and verify that the signature attached to the block is valid
 // for the certificate provided in the block.
 func Parse(bytes []byte, chainID ids.ID) (Block, error) {

vms/proposervm/block/parse_test.go

@@ -14,8 +14,69 @@ import (
 	"github.com/ava-labs/avalanchego/codec"
 	"github.com/ava-labs/avalanchego/ids"
 	"github.com/ava-labs/avalanchego/staking"
+	"github.com/ava-labs/avalanchego/utils/wrappers"
 )
 
+func TestParseBlocks(t *testing.T) {
+	parentID := ids.ID{1}
+	timestamp := time.Unix(123, 0)
+	pChainHeight := uint64(2)
+	innerBlockBytes := []byte{3}
+	chainID := ids.ID{4}
+
+	tlsCert, err := staking.NewTLSCert()
+	require.NoError(t, err)
+
+	cert, err := staking.ParseCertificate(tlsCert.Leaf.Raw)
+	require.NoError(t, err)
+	key := tlsCert.PrivateKey.(crypto.Signer)
+
+	signedBlock, err := Build(
+		parentID,
+		timestamp,
+		pChainHeight,
+		cert,
+		innerBlockBytes,
+		chainID,
+		key,
+	)
+	require.NoError(t, err)
+
+	signedBlockBytes := signedBlock.Bytes()
+	malformedBlockBytes := make([]byte, len(signedBlockBytes)-1)
+	copy(malformedBlockBytes, signedBlockBytes)
+
+	for _, testCase := range []struct {
+		name   string
+		input  [][]byte
+		output []ParseResult
+	}{
+		{
+			name:   "ValidThenInvalid",
+			input:  [][]byte{signedBlockBytes, malformedBlockBytes},
+			output: []ParseResult{{Block: &statelessBlock{bytes: signedBlockBytes}}, {Err: wrappers.ErrInsufficientLength}},
+		},
+		{
+			name:   "InvalidThenValid",
+			input:  [][]byte{malformedBlockBytes, signedBlockBytes},
+			output: []ParseResult{{Err: wrappers.ErrInsufficientLength}, {Block: &statelessBlock{bytes: signedBlockBytes}}},
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			results := ParseBlocks(testCase.input, chainID)
+			for i := range testCase.output {
+				if testCase.output[i].Block == nil {
+					require.Nil(t, results[i].Block)
+					require.ErrorIs(t, results[i].Err, testCase.output[i].Err)
+				} else {
+					require.Equal(t, testCase.output[i].Block.Bytes(), results[i].Block.Bytes())
+					require.NoError(t, results[i].Err)
+				}
+			}
+		})
+	}
+}
+
 func TestParse(t *testing.T) {
 	parentID := ids.ID{1}
 	timestamp := time.Unix(123, 0)