Parallelize BatchedParseBlock

Make BatchedParseBlock parse blocks (and verify their signature) in parallel.

Prior to this code change, batch parsing of blocks in the proposer VM would only work if the wrapped VMs support batch processing.
Now, if the wrapped VM doesn't support batch processing, it simply parses them one by one, but the proposer VM still parses in parallel.

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,9 @@ package proposervm
 import (
 	"bytes"
 	"context"
+	"crypto"
+	"encoding/binary"
+	"reflect"
 	"testing"
 	"time"
 
@@ -23,7 +26,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 +58,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 +590,105 @@ func TestBatchedParseBlockPreForkOnly(t *testing.T) {
 	require.Equal(builtBlk3.ID(), res[2].ID())
 }
 
+func TestBatchedParseBlockParallel(t *testing.T) {
+	req := require.New(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()
+	req.NoError(err)
+
+	cert, err := staking.ParseCertificate(tlsCert.Leaf.Raw)
+	req.NoError(err)
+	key := tlsCert.PrivateKey.(crypto.Signer)
+
+	blockThatCantBeParsed := snowmantest.BuildChild(snowmantest.Genesis)
+
+	blockBatch1 := makeParseAbleBlocks(parentID, timestamp, pChainHeight, cert, chainID, key, req)
+	blockBatch1[50] = blockThatCantBeParsed.Bytes()
+
+	blockBatch2 := makeParseAbleBlocks(parentID, timestamp, pChainHeight, cert, chainID, key, req)
+
+	for _, testCase := range []struct {
+		name         string
+		preForkIndex int
+		rawBlocks    [][]byte
+	}{
+		{
+			name: "empty input",
+		},
+		{
+			name:         "pre-fork is somewhere in the middle",
+			rawBlocks:    blockBatch1,
+			preForkIndex: 50,
+		},
+		{
+			name:         "all blocks are post fork",
+			rawBlocks:    blockBatch2,
+			preForkIndex: 9999,
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			require := require.New(t)
+			blocks, err := vm.BatchedParseBlock(context.Background(), testCase.rawBlocks)
+			require.NoError(err)
+
+			var expectedBlocks [][]byte
+			for _, block := range blocks {
+				expectedBlocks = append(expectedBlocks, block.Bytes())
+			}
+
+			require.Equal(expectedBlocks, testCase.rawBlocks)
+
+			for i, block := range blocks {
+				if i < testCase.preForkIndex {
+					require.Equal(reflect.TypeOf(block), reflect.TypeOf(&postForkBlock{}))
+				} else {
+					require.Equal(reflect.TypeOf(block), reflect.TypeOf(&preForkBlock{}))
+				}
+			}
+		})
+	}
+}
+
+func makeParseAbleBlocks(parentID ids.ID, timestamp time.Time, pChainHeight uint64, cert *staking.Certificate, chainID ids.ID, key crypto.Signer, req *require.Assertions) [][]byte {
+	makeSignedBlock := func(i int) []byte {
+		buff := make([]byte, 2)
+		binary.BigEndian.PutUint16(buff, uint16(i))
+
+		signedBlock, err := blockbuilder.Build(
+			parentID,
+			timestamp,
+			pChainHeight,
+			cert,
+			buff,
+			chainID,
+			key,
+		)
+		req.NoError(err)
+
+		return signedBlock.Bytes()
+	}
+
+	var blockBatch1 [][]byte
+	for i := 0; i < 100; i++ {
+		blockBatch1 = append(blockBatch1, makeSignedBlock(i))
+	}
+	return blockBatch1
+}
+
 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 would parse the given raw blocks into tuples of (Block, error).
+// Each ParseResult in the output is returned in the same order as its corresponding raw block 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

@@ -9,13 +9,58 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ava-labs/avalanchego/utils/wrappers"
+
 	"github.com/stretchr/testify/require"
 
 	"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)
+
+	results := ParseBlocks([][]byte{signedBlockBytes, malformedBlockBytes}, chainID)
+	require.Equal(t, results[0].Block.Bytes(), signedBlockBytes)
+	require.NoError(t, results[0].Err)
+	require.ErrorIs(t, results[1].Err, wrappers.ErrInsufficientLength)
+	require.Nil(t, results[1].Block)
+
+	results = ParseBlocks([][]byte{malformedBlockBytes, signedBlockBytes}, chainID)
+	require.Equal(t, results[1].Block.Bytes(), signedBlockBytes)
+	require.NoError(t, results[1].Err)
+	require.ErrorIs(t, results[0].Err, wrappers.ErrInsufficientLength)
+	require.Nil(t, results[0].Block)
+}
+
 func TestParse(t *testing.T) {
 	parentID := ids.ID{1}
 	timestamp := time.Unix(123, 0)