Benchmark framework for evaluating ledger performance

https://jira.hyperledger.org/browse/FAB-1309

 - Basic framework for creating and managing chains
 - Basic framework for transaction submission, block creation, and block commit
 - Benchmark tests for insert and read-write transactions
 - scripts for common functions
 - scripts for launching the benchmarks

Change-Id: I03f09fa2eec4fde5a9b23fc2316718c9a33577b5
Signed-off-by: manish <manish.sethi@gmail.com>

common/ledger/testutil/test_helper.go

@@ -21,15 +21,12 @@ import (
 
 	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/common/configtx/test"
-	"github.com/hyperledger/fabric/common/configtx/tool/provisional"
 	"github.com/hyperledger/fabric/common/util"
 	lutils "github.com/hyperledger/fabric/core/ledger/util"
 	"github.com/hyperledger/fabric/protos/common"
 	pb "github.com/hyperledger/fabric/protos/peer"
 	ptestutils "github.com/hyperledger/fabric/protos/testutils"
 	"github.com/hyperledger/fabric/protos/utils"
-
-	genesisconfig "github.com/hyperledger/fabric/common/configtx/tool/localconfig"
 )
 
 //BlockGenerator generates a series of blocks for testing
@@ -147,9 +144,3 @@ func newBlock(env []*common.Envelope, blockNum uint64, previousHash []byte) *com
 
 	return block
 }
-
-func MakeGenesisBlock() *common.Block {
-	genConf := genesisconfig.Load(genesisconfig.SampleInsecureProfile)
-	gb := provisional.New(genConf).GenesisBlock()
-	return gb
-}

test/tools/ledgerbenchmarks/chainmgmt/block_gen.go

@@ -0,0 +1,106 @@
+/*
+Copyright IBM Corp. 2017 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+		 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package chainmgmt
+
+import (
+	"sync"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/protos/common"
+	benchcommon "github.com/hyperledger/fabric/test/tools/ledgerbenchmarks/common"
+)
+
+const (
+	numConcurrentTxEnvCreators = 30
+)
+
+type txEnvBytes []byte
+
+// blkGenerator generates blocks in sequence. One instance of blkGenerator is maintained for each chain
+type blkGenerator struct {
+	batchConf         *BatchConf
+	blockNum          uint64
+	previousBlockHash []byte
+
+	srQueue chan SimulationResult
+	txQueue chan txEnvBytes
+	wg      *sync.WaitGroup
+}
+
+func newBlkGenerator(batchConf *BatchConf, startingBlockNum uint64, previousBlockHash []byte) *blkGenerator {
+	bg := &blkGenerator{
+		batchConf,
+		startingBlockNum,
+		previousBlockHash,
+		make(chan SimulationResult, batchConf.BatchSize),
+		make(chan txEnvBytes, batchConf.BatchSize),
+		&sync.WaitGroup{},
+	}
+	bg.startTxEnvCreators()
+	return bg
+}
+
+func (bg *blkGenerator) startTxEnvCreators() {
+	for i := 0; i < numConcurrentTxEnvCreators; i++ {
+		go bg.startTxEnvCreator()
+	}
+}
+
+func (bg *blkGenerator) startTxEnvCreator() {
+	bg.wg.Add(1)
+	for sr := range bg.srQueue {
+		txEnv, err := createTxEnv(sr)
+		benchcommon.PanicOnError(err)
+		txEnvBytes, err := proto.Marshal(txEnv)
+		benchcommon.PanicOnError(err)
+		bg.txQueue <- txEnvBytes
+	}
+	bg.wg.Done()
+}
+
+func (bg *blkGenerator) addTx(sr SimulationResult) {
+	bg.srQueue <- sr
+}
+
+func (bg *blkGenerator) nextBlock() *common.Block {
+	block := common.NewBlock(bg.blockNum, bg.previousBlockHash)
+	numTx := 0
+	for txEnvBytes := range bg.txQueue {
+		numTx++
+		block.Data.Data = append(block.Data.Data, txEnvBytes)
+		if numTx == bg.batchConf.BatchSize {
+			break
+		}
+	}
+	// close() has been called and no pending tx
+	if len(block.Data.Data) == 0 {
+		return nil
+	}
+	block.Header.DataHash = block.Data.Hash()
+	block.Header.Number = bg.blockNum
+	block.Header.PreviousHash = bg.previousBlockHash
+
+	bg.blockNum++
+	bg.previousBlockHash = block.Header.Hash()
+	return block
+}
+
+func (bg *blkGenerator) close() {
+	close(bg.srQueue)
+	bg.wg.Wait()
+	close(bg.txQueue)
+}

test/tools/ledgerbenchmarks/chainmgmt/chains.go

@@ -0,0 +1,142 @@
+/*
+Copyright IBM Corp. 2017 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+		 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package chainmgmt
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/hyperledger/fabric/common/configtx/test"
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/ledgermgmt"
+	"github.com/hyperledger/fabric/protos/common"
+	benchcommon "github.com/hyperledger/fabric/test/tools/ledgerbenchmarks/common"
+)
+
+// ChainID is a type used for the ids for the chains for experiments
+type ChainID int
+
+func (chainID ChainID) String() string {
+	return fmt.Sprintf("%s%04d", "chain_", chainID)
+}
+
+// SimulationResult is a type used for simulation results
+type SimulationResult []byte
+
+// chainsMgr manages chains for experiments
+type chainsMgr struct {
+	mgrConf   *ChainMgrConf
+	batchConf *BatchConf
+	initOp    chainInitOp
+	chainsMap map[ChainID]*Chain
+	wg        *sync.WaitGroup
+}
+
+func newChainsMgr(mgrConf *ChainMgrConf, batchConf *BatchConf, initOp chainInitOp) *chainsMgr {
+	ledgermgmt.Initialize()
+	return &chainsMgr{mgrConf, batchConf, initOp, make(map[ChainID]*Chain), &sync.WaitGroup{}}
+}
+
+func (m *chainsMgr) createOrOpenChains() []*Chain {
+	var ledgerInitFunc func(string) (ledger.PeerLedger, error)
+	switch m.initOp {
+	case ChainInitOpCreate:
+		ledgerInitFunc = createLedgerByID
+	case ChainInitOpOpen:
+		ledgerInitFunc = ledgermgmt.OpenLedger
+	default:
+		panic(fmt.Errorf("unknown chain init opeartion"))
+	}
+
+	numChains := m.mgrConf.NumChains
+	for i := 0; i < numChains; i++ {
+		chainID := ChainID(i)
+		peerLedger, err := ledgerInitFunc(chainID.String())
+		benchcommon.PanicOnError(err)
+		c := newChain(chainID, peerLedger, m)
+		m.chainsMap[chainID] = c
+	}
+	return m.chains()
+}
+
+func (m *chainsMgr) chains() []*Chain {
+	chains := []*Chain{}
+	for _, chain := range m.chainsMap {
+		chains = append(chains, chain)
+	}
+	return chains
+}
+
+func (m *chainsMgr) waitForChainsToExhaustAllBlocks() {
+	m.wg.Wait()
+	ledgermgmt.Close()
+}
+
+// Chain extends ledger.PeerLedger and the experiments invoke ledger functions via a chain type
+type Chain struct {
+	ledger.PeerLedger
+	ID           ChainID
+	blkGenerator *blkGenerator
+	m            *chainsMgr
+}
+
+func newChain(id ChainID, peerLedger ledger.PeerLedger, m *chainsMgr) *Chain {
+	bcInfo, err := peerLedger.GetBlockchainInfo()
+	benchcommon.PanicOnError(err)
+	return &Chain{peerLedger, id, newBlkGenerator(m.batchConf, bcInfo.Height, bcInfo.CurrentBlockHash), m}
+}
+
+func (c *Chain) startBlockPollingAndCommit() {
+	c.m.wg.Add(1)
+	go func() {
+		defer c.close()
+		for {
+			block := c.blkGenerator.nextBlock()
+			if block == nil {
+				break
+			}
+			benchcommon.PanicOnError(c.PeerLedger.Commit(block))
+		}
+	}()
+}
+
+// SubmitTx is expected to be called by an experiment for submitting the transactions
+func (c *Chain) SubmitTx(sr SimulationResult) {
+	c.blkGenerator.addTx(sr)
+}
+
+// Done is expected to be called by an experiment when the experiment does not have any more transactions to submit
+func (c *Chain) Done() {
+	c.blkGenerator.close()
+}
+
+// Commit overrides the Commit function in ledger.PeerLedger because,
+// experiments are not expected to call Commit directly to the ledger
+func (c *Chain) Commit(block *common.Block) {
+	panic(fmt.Errorf("Commit should not be invoked directly"))
+}
+
+func (c *Chain) close() {
+	c.PeerLedger.Close()
+	c.m.wg.Done()
+}
+
+func createLedgerByID(ledgerid string) (ledger.PeerLedger, error) {
+	gb, err := test.MakeGenesisBlock(ledgerid)
+	benchcommon.PanicOnError(err)
+	return ledgermgmt.CreateLedger(gb)
+}

test/tools/ledgerbenchmarks/chainmgmt/conf.go

@@ -0,0 +1,33 @@
+/*
+Copyright IBM Corp. 2017 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+		 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package chainmgmt
+
+// ChainMgrConf captures the configurations meant at the level of chainMgr
+// DataDir field specifies the filesystem location where the chains data is maintained
+// NumChains field specifies the number of chains to instantiate
+type ChainMgrConf struct {
+	DataDir   string
+	NumChains int
+}
+
+// BatchConf captures the batch related configurations
+// BatchSize specifies the number of transactions in one block
+// SignBlock specifies whether the transactions should be signed
+type BatchConf struct {
+	BatchSize int
+	SignBlock bool
+}

test/tools/ledgerbenchmarks/chainmgmt/testenv.go

@@ -0,0 +1,62 @@
+/*
+Copyright IBM Corp. 2017 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+		 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package chainmgmt
+
+import "github.com/spf13/viper"
+
+// chainInitOp is a type that an experiment uses to specify how the chains
+// should be initialized at the beginning of the experiment. See below the
+// enum values for this type
+type chainInitOp uint8
+
+const (
+	// ChainInitOpCreate indicates that the chains should be creates afresh
+	ChainInitOpCreate chainInitOp = iota + 1
+	// ChainInitOpOpen indicates that the existing chains should be opened
+	ChainInitOpOpen
+)
+
+// TestEnv is a high level struct that the experiments are expeted to use as a starting point.
+// See one of the Benchmark tests for the intented usage
+type TestEnv struct {
+	mgr *chainsMgr
+}
+
+// InitTestEnv initialize TestEnv with given configurations. The initialization cuases
+// creation (or openning of existing) chains and the block creation and commit go routines
+// for each of the chains. For configurations options, see comments on specific configuration type
+func InitTestEnv(mgrConf *ChainMgrConf, batchConf *BatchConf, initOperation chainInitOp) *TestEnv {
+	viper.Set("peer.fileSystemPath", mgrConf.DataDir)
+	mgr := newChainsMgr(mgrConf, batchConf, initOperation)
+	chains := mgr.createOrOpenChains()
+	for _, chain := range chains {
+		chain.startBlockPollingAndCommit()
+	}
+	return &TestEnv{mgr}
+}
+
+// Chains returns handle to all the chains
+func (env TestEnv) Chains() []*Chain {
+	return env.mgr.chains()
+}
+
+// WaitForTestCompletion waits till all the transactions are committed
+// An experiment after launching all the goroutine should call this
+// so that the process is alive till all the goroutines complete
+func (env TestEnv) WaitForTestCompletion() {
+	env.mgr.waitForChainsToExhaustAllBlocks()
+}