Merge "[FAB-5932] - Parallel tx validation"

core/chaincode/ccproviderimpl.go

@@ -43,7 +43,6 @@ func init() {
 
 // ccProviderImpl is an implementation of the ccprovider.ChaincodeProvider interface
 type ccProviderImpl struct {
-	txsim ledger.TxSimulator
 }
 
 // ccProviderContextImpl contains the state that is passed around to calls to methods of ccProviderImpl
@@ -52,15 +51,15 @@ type ccProviderContextImpl struct {
 }
 
 // GetContext returns a context for the supplied ledger, with the appropriate tx simulator
-func (c *ccProviderImpl) GetContext(ledger ledger.PeerLedger, txid string) (context.Context, error) {
+func (c *ccProviderImpl) GetContext(ledger ledger.PeerLedger, txid string) (context.Context, ledger.TxSimulator, error) {
 	var err error
 	// get context for the chaincode execution
-	c.txsim, err = ledger.NewTxSimulator(txid)
+	txsim, err := ledger.NewTxSimulator(txid)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	ctxt := context.WithValue(context.Background(), TXSimulatorKey, c.txsim)
-	return ctxt, nil
+	ctxt := context.WithValue(context.Background(), TXSimulatorKey, txsim)
+	return ctxt, txsim, nil
 }
 
 // GetCCContext returns an interface that encapsulates a
@@ -114,8 +113,3 @@ func (c *ccProviderImpl) Stop(ctxt context.Context, cccid interface{}, spec *pb.
 	}
 	panic("ChaincodeSupport not initialized")
 }
-
-// ReleaseContext frees up resources held by the context
-func (c *ccProviderImpl) ReleaseContext() {
-	c.txsim.Done()
-}

core/committer/txvalidator/txvalidator_test.go

@@ -38,18 +38,10 @@ import (
 	"github.com/hyperledger/fabric/protos/utils"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
+	"golang.org/x/sync/semaphore"
 )
 
-func TestBlockValidation(t *testing.T) {
-	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
-	ledgermgmt.InitializeTestEnv()
-	defer ledgermgmt.CleanupTestEnv()
-
-	gb, _ := test.MakeGenesisBlock("TestLedger")
-	gbHash := gb.Header.Hash()
-	ledger, _ := ledgermgmt.CreateLedger(gb)
-	defer ledger.Close()
-
+func testValidationWithNTXes(t *testing.T, ledger ledger2.PeerLedger, gbHash []byte, nBlocks int) {
 	txid := util2.GenerateUUID()
 	simulator, _ := ledger.NewTxSimulator(txid)
 	simulator.SetState("ns1", "key1", []byte("value1"))
@@ -65,18 +57,29 @@ func TestBlockValidation(t *testing.T) {
 	}
 
 	mockVsccValidator := &validator.MockVsccValidator{}
-	tValidator := &txValidator{&mocktxvalidator.Support{LedgerVal: ledger}, mockVsccValidator}
+	vcs := struct {
+		*mocktxvalidator.Support
+		*semaphore.Weighted
+	}{&mocktxvalidator.Support{LedgerVal: ledger}, semaphore.NewWeighted(10)}
+	tValidator := &txValidator{vcs, mockVsccValidator}
 
 	bcInfo, _ := ledger.GetBlockchainInfo()
 	testutil.AssertEquals(t, bcInfo, &common.BlockchainInfo{
 		Height: 1, CurrentBlockHash: gbHash, PreviousBlockHash: nil})
 
-	block := testutil.ConstructBlock(t, 1, gbHash, [][]byte{pubSimulationResBytes}, true)
+	sr := [][]byte{}
+	for i := 0; i < nBlocks; i++ {
+		sr = append(sr, pubSimulationResBytes)
+	}
+	block := testutil.ConstructBlock(t, 1, gbHash, sr, true)
 
 	tValidator.Validate(block)
 
 	txsfltr := util.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
-	assert.True(t, txsfltr.IsSetTo(0, peer.TxValidationCode_VALID))
+
+	for i := 0; i < nBlocks; i++ {
+		assert.True(t, txsfltr.IsSetTo(i, peer.TxValidationCode_VALID))
+	}
 
 	/*
 
@@ -100,6 +103,50 @@ func TestBlockValidation(t *testing.T) {
 	*/
 }
 
+func TestBlockValidation(t *testing.T) {
+	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
+	ledgermgmt.InitializeTestEnv()
+	defer ledgermgmt.CleanupTestEnv()
+
+	gb, _ := test.MakeGenesisBlock("TestLedger")
+	gbHash := gb.Header.Hash()
+	ledger, _ := ledgermgmt.CreateLedger(gb)
+	defer ledger.Close()
+
+	// here we test validation of a block with a single tx
+	testValidationWithNTXes(t, ledger, gbHash, 1)
+}
+
+func TestParallelBlockValidation(t *testing.T) {
+	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
+	ledgermgmt.InitializeTestEnv()
+	defer ledgermgmt.CleanupTestEnv()
+
+	gb, _ := test.MakeGenesisBlock("TestLedger")
+	gbHash := gb.Header.Hash()
+	ledger, _ := ledgermgmt.CreateLedger(gb)
+	defer ledger.Close()
+
+	// here we test validation of a block with 128 txes
+	testValidationWithNTXes(t, ledger, gbHash, 128)
+}
+
+func TestVeryLargeParallelBlockValidation(t *testing.T) {
+	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
+	ledgermgmt.InitializeTestEnv()
+	defer ledgermgmt.CleanupTestEnv()
+
+	gb, _ := test.MakeGenesisBlock("TestLedger")
+	gbHash := gb.Header.Hash()
+	ledger, _ := ledgermgmt.CreateLedger(gb)
+	defer ledger.Close()
+
+	// here we test validation of a block with 4096 txes,
+	// which is larger than both the number of workers in
+	// the pool and the buffer in the channels
+	testValidationWithNTXes(t, ledger, gbHash, 4096)
+}
+
 func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
 	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
 	ledgermgmt.InitializeTestEnv()
@@ -110,7 +157,11 @@ func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
 
 	defer ledger.Close()
 
-	tValidator := &txValidator{&mocktxvalidator.Support{LedgerVal: ledger}, &validator.MockVsccValidator{}}
+	vcs := struct {
+		*mocktxvalidator.Support
+		*semaphore.Weighted
+	}{&mocktxvalidator.Support{LedgerVal: ledger}, semaphore.NewWeighted(10)}
+	tValidator := &txValidator{vcs, &validator.MockVsccValidator{}}
 
 	// Create simple endorsement transaction
 	payload := &common.Payload{