Merge branch 'master' into sync

benchmark.go

@@ -50,6 +50,7 @@ func main() {
 	attackedMode := flag.Int("attacked_mode", 0, "0 means not attacked, 1 means attacked, 2 means being open to be selected as attacked")
 	dbSupported := flag.Bool("db_supported", false, "false means not db_supported, true means db_supported")
 	profile := flag.Bool("profile", false, "Turn on profiling (CPU, Memory).")
+	metricsReportURL := flag.String("metrics_profile_url", "", "If set, reports metrics to this URL.")
 	flag.Parse()
 
 	// Set up randomization seed.
@@ -90,9 +91,12 @@ func main() {
 	consensus := consensus.NewConsensus(*ip, *port, shardID, peers, leader)
 
 	// Start Profiler for leader if profile argument is on
-	if *profile && role == "leader" {
-		profiler := profiler.NewProfiler(consensus.Log, os.Getpid(), shardID)
-		profiler.Start()
+	if role == "leader" && (*profile || *metricsReportURL != "") {
+		prof := profiler.GetProfiler()
+		prof.Config(consensus.Log, shardID, *metricsReportURL)
+		if *profile {
+			prof.Start()
+		}
 	}
 
 	// Set logger to attack model.

blockchain/utxopool.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/gob"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"github.com/dedis/kyber/sign/schnorr"
 	"github.com/simple-rules/harmony-benchmark/crypto"
@@ -77,7 +78,7 @@ func (utxoPool *UTXOPool) VerifyTransactions(transactions []*Transaction) bool {
 	spentTXOs := make(map[[20]byte]map[string]map[uint32]bool)
 	if utxoPool != nil {
 		for _, tx := range transactions {
-			if valid, crossShard := utxoPool.VerifyOneTransaction(tx, &spentTXOs); !crossShard && !valid {
+			if err, crossShard := utxoPool.VerifyOneTransaction(tx, &spentTXOs); !crossShard && err != nil {
 				return false
 			}
 		}
@@ -114,7 +115,7 @@ func (utxoPool *UTXOPool) VerifyStateBlock(stateBlock *Block) bool {
 }
 
 // VerifyOneTransaction verifies if a list of transactions valid.
-func (utxoPool *UTXOPool) VerifyOneTransaction(tx *Transaction, spentTXOs *map[[20]byte]map[string]map[uint32]bool) (valid, crossShard bool) {
+func (utxoPool *UTXOPool) VerifyOneTransaction(tx *Transaction, spentTXOs *map[[20]byte]map[string]map[uint32]bool) (err error, crossShard bool) {
 	if len(tx.Proofs) != 0 {
 		return utxoPool.VerifyUnlockTransaction(tx)
 	}
@@ -133,10 +134,10 @@ func (utxoPool *UTXOPool) VerifyOneTransaction(tx *Transaction, spentTXOs *map[[
 
 		inTxID := hex.EncodeToString(in.PreviousOutPoint.TxID[:])
 		index := in.PreviousOutPoint.Index
-		// Check if the transaction with the addres is spent or not.
+		// Check if the transaction with the address is spent or not.
 		if val, ok := (*spentTXOs)[in.Address][inTxID][index]; ok {
 			if val {
-				return false, crossShard
+				return errors.New("TxInput is already spent"), crossShard
 			}
 		}
 		// Mark the transactions with the address and index spent.
@@ -154,7 +155,7 @@ func (utxoPool *UTXOPool) VerifyOneTransaction(tx *Transaction, spentTXOs *map[[
 			inTotal += val
 		} else {
 			utxoPool.mutex.Unlock()
-			return false, crossShard
+			return errors.New("Specified TxInput does not exist in utxo pool"), crossShard
 		}
 		utxoPool.mutex.Unlock()
 	}
@@ -171,30 +172,30 @@ func (utxoPool *UTXOPool) VerifyOneTransaction(tx *Transaction, spentTXOs *map[[
 
 	// TODO: improve this checking logic
 	if (crossShard && inTotal > outTotal) || (!crossShard && inTotal != outTotal) {
-		return false, crossShard
+		return errors.New("Input and output amount doesn't match"), crossShard
 	}
 
 	if inTotal == 0 {
-		return false, false // Here crossShard is false, because if there is no business for this shard, it's effectively not crossShard no matter what.
+		return errors.New("Input amount is 0"), false // Here crossShard is false, because if there is no business for this shard, it's effectively not crossShard no matter what.
 	}
 
 	// Verify the signature
 	pubKey := crypto.Ed25519Curve.Point()
-	err := pubKey.UnmarshalBinary(tx.PublicKey[:])
-	if err != nil {
-		log.Error("Failed to deserialize public key", "error", err)
+	tempErr := pubKey.UnmarshalBinary(tx.PublicKey[:])
+	if tempErr != nil {
+		log.Error("Failed to deserialize public key", "error", tempErr)
 	}
-	err = schnorr.Verify(crypto.Ed25519Curve, pubKey, tx.GetContentToVerify(), tx.Signature[:])
-	if err != nil {
-		log.Error("Failed to verify signature", "error", err, "public key", pubKey, "pubKey in bytes", tx.PublicKey[:])
-		return false, crossShard
+	tempErr = schnorr.Verify(crypto.Ed25519Curve, pubKey, tx.GetContentToVerify(), tx.Signature[:])
+	if tempErr != nil {
+		log.Error("Failed to verify signature", "error", tempErr, "public key", pubKey, "pubKey in bytes", tx.PublicKey[:])
+		return errors.New("Invalid signature"), crossShard
 	}
-	return true, crossShard
+	return nil, crossShard
 }
 
 // Verify a cross shard transaction that contains proofs for unlock-to-commit/abort.
-func (utxoPool *UTXOPool) VerifyUnlockTransaction(tx *Transaction) (valid, crossShard bool) {
-	valid = true
+func (utxoPool *UTXOPool) VerifyUnlockTransaction(tx *Transaction) (err error, crossShard bool) {
+	err = nil
 	crossShard = false // unlock transaction is treated as crossShard=false because it will be finalized now (doesn't need more steps)
 	txInputs := make(map[TXInput]bool)
 	for _, curProof := range tx.Proofs {
@@ -205,7 +206,7 @@ func (utxoPool *UTXOPool) VerifyUnlockTransaction(tx *Transaction) (valid, cross
 	for _, txInput := range tx.TxInput {
 		val, ok := txInputs[txInput]
 		if !ok || !val {
-			valid = false
+			err = errors.New("Invalid unlock transaction: not all proofs are provided for tx inputs")
 		}
 	}
 	return
@@ -346,7 +347,7 @@ func (utxoPool *UTXOPool) UpdateOneTransaction(tx *Transaction) {
 // VerifyOneTransactionAndUpdate verifies and update a valid transaction.
 // Return false if the transaction is not valid.
 func (utxoPool *UTXOPool) VerifyOneTransactionAndUpdate(tx *Transaction) bool {
-	if valid, _ := utxoPool.VerifyOneTransaction(tx, nil); valid {
+	if err, _ := utxoPool.VerifyOneTransaction(tx, nil); err == nil {
 		utxoPool.UpdateOneTransaction(tx)
 		return true
 	}
@@ -385,29 +386,28 @@ func CreateUTXOPoolFromGenesisBlockChain(bc *Blockchain) *UTXOPool {
 }
 
 // SelectTransactionsForNewBlock returns a list of index of valid transactions for the new block.
-func (utxoPool *UTXOPool) SelectTransactionsForNewBlock(transactions []*Transaction, maxNumTxs int) ([]*Transaction, []*Transaction, []*CrossShardTxAndProof) {
-	selected, unselected, crossShardTxs := []*Transaction{}, []*Transaction{}, []*CrossShardTxAndProof{}
+func (utxoPool *UTXOPool) SelectTransactionsForNewBlock(transactions []*Transaction, maxNumTxs int) ([]*Transaction, []*Transaction, []*Transaction, []*CrossShardTxAndProof) {
+	selected, unselected, invalid, crossShardTxs := []*Transaction{}, []*Transaction{}, []*Transaction{}, []*CrossShardTxAndProof{}
 	spentTXOs := make(map[[20]byte]map[string]map[uint32]bool)
 	for _, tx := range transactions {
-		valid, crossShard := utxoPool.VerifyOneTransaction(tx, &spentTXOs)
+		err, crossShard := utxoPool.VerifyOneTransaction(tx, &spentTXOs)
 
 		if len(selected) < maxNumTxs {
-			if valid || crossShard {
+			if err == nil || crossShard {
 				selected = append(selected, tx)
 				if crossShard {
-					proof := CrossShardTxProof{Accept: valid, TxID: tx.ID, TxInput: getShardTxInput(tx, utxoPool.ShardID)}
+					proof := CrossShardTxProof{Accept: err == nil, TxID: tx.ID, TxInput: getShardTxInput(tx, utxoPool.ShardID)}
 					txAndProof := CrossShardTxAndProof{tx, &proof}
 					crossShardTxs = append(crossShardTxs, &txAndProof)
 				}
 			} else {
-				unselected = append(unselected, tx)
+				invalid = append(invalid, tx)
 			}
 		} else {
-			// TODO: discard invalid transactions
 			unselected = append(unselected, tx)
 		}
 	}
-	return selected, unselected, crossShardTxs
+	return selected, unselected, invalid, crossShardTxs
 }
 
 func getShardTxInput(transaction *Transaction, shardID uint32) []TXInput {

blockchain/utxopool_test.go

@@ -16,7 +16,7 @@ func TestVerifyOneTransactionAndUpdate(t *testing.T) {
 		t.Error("failed to create a new transaction.")
 	}
 
-	if valid, _ := utxoPool.VerifyOneTransaction(tx, nil); !valid {
+	if err, _ := utxoPool.VerifyOneTransaction(tx, nil); err != nil {
 		t.Error("failed to verify a valid transaction.")
 	}
 	utxoPool.VerifyOneTransactionAndUpdate(tx)
@@ -35,7 +35,7 @@ func TestVerifyOneTransactionFail(t *testing.T) {
 	}
 
 	tx.TxInput = append(tx.TxInput, tx.TxInput[0])
-	if valid, _ := utxoPool.VerifyOneTransaction(tx, nil); valid {
+	if err, _ := utxoPool.VerifyOneTransaction(tx, nil); err == nil {
 		t.Error("Tx with multiple identical TxInput shouldn't be valid")
 	}
 }

client/txgen/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"encoding/binary"
 	"encoding/hex"
 	"flag"
 	"fmt"
@@ -57,12 +58,13 @@ type TxInfo struct {
 //       token (1000) to each address in [0 - N). See node.AddTestingAddresses()
 //
 // Params:
+//     subsetId                   - the which subset of the utxo to work on (used to select addresses)
 //     shardID                    - the shardID for current shard
 //     dataNodes                  - nodes containing utxopools of all shards
 // Returns:
 //     all single-shard txs
 //     all cross-shard txs
-func generateSimulatedTransactions(shardID int, dataNodes []*node.Node) ([]*blockchain.Transaction, []*blockchain.Transaction) {
+func generateSimulatedTransactions(subsetId, numSubset int, shardID int, dataNodes []*node.Node) ([]*blockchain.Transaction, []*blockchain.Transaction) {
 	/*
 	  UTXO map structure:
 	     address - [
@@ -86,6 +88,7 @@ func generateSimulatedTransactions(shardID int, dataNodes []*node.Node) ([]*bloc
 UTXOLOOP:
 	// Loop over all addresses
 	for address, txMap := range dataNodes[shardID].UtxoPool.UtxoMap {
+		if int(binary.BigEndian.Uint32(address[:]))%numSubset == subsetId%numSubset { // Work on one subset of utxo at a time
 			txInfo.address = address
 			// Loop over all txIds for the address
 			for txIdStr, utxoMap := range txMap {
@@ -102,11 +105,8 @@ UTXOLOOP:
 					txInfo.value = value
 
 					randNum := rand.Intn(100)
-				// 30% sample rate to select UTXO to use for new transactions
-				if randNum >= 30 {
-					continue
-				}
-				if setting.crossShard && randNum < 10 { // 1/3 cross shard transactions: add another txinput from another shard
+
+					if setting.crossShard && randNum < 30 { // 1/3 cross shard transactions: add another txinput from another shard
 						generateCrossShardTx(&txInfo)
 					} else {
 						generateSingleShardTx(&txInfo)
@@ -117,6 +117,7 @@ UTXOLOOP:
 				}
 			}
 		}
+	}
 	utxoPoolMutex.Unlock()
 
 	log.Debug("[Generator] generated transations", "single-shard", len(txInfo.txs), "cross-shard", len(txInfo.crossTxs))
@@ -125,8 +126,8 @@ UTXOLOOP:
 
 func generateCrossShardTx(txInfo *TxInfo) {
 	nodeShardID := txInfo.dataNodes[txInfo.shardID].Consensus.ShardID
-	// shard with neighboring Id
-	crossShardId := (int(nodeShardID) + 1) % len(txInfo.dataNodes)
+	// a random shard to spend money to
+	crossShardId := rand.Intn(len(txInfo.dataNodes))
 
 	crossShardNode := txInfo.dataNodes[crossShardId]
 	crossShardUtxosMap := crossShardNode.UtxoPool.UtxoMap[txInfo.address]
@@ -241,6 +242,8 @@ func main() {
 	configFile := flag.String("config_file", "local_config.txt", "file containing all ip addresses and config")
 	maxNumTxsPerBatch := flag.Int("max_num_txs_per_batch", 100000, "number of transactions to send per message")
 	logFolder := flag.String("log_folder", "latest", "the folder collecting the logs of this execution")
+	numSubset := flag.Int("numSubset", 3, "the number of subsets of utxos to process separately")
+	duration := flag.Int("duration", 60, "duration of the tx generation in second")
 	flag.Parse()
 
 	// Read the configs
@@ -289,6 +292,7 @@ func main() {
 						// Add it to blockchain
 						utxoPoolMutex.Lock()
 						node.AddNewBlock(block)
+						node.UpdateUtxoAndState(block)
 						utxoPoolMutex.Unlock()
 					} else {
 						continue
@@ -302,32 +306,45 @@ func main() {
 		go func() {
 			clientNode.StartServer(clientPort)
 		}()
-
 	}
 
 	// Transaction generation process
 	time.Sleep(10 * time.Second) // wait for nodes to be ready
 	start := time.Now()
-	totalTime := 60.0 //run for 1 minutes
+	totalTime := float64(*duration)
 
+	batchCounter := 0
 	for true {
 		t := time.Now()
 		if t.Sub(start).Seconds() >= totalTime {
 			log.Debug("Generator timer ended.", "duration", (int(t.Sub(start))), "startTime", start, "totalTime", totalTime)
 			break
 		}
+		for shardId := 0; shardId < len(nodes); shardId++ {
+			constructAndSendTransaction(batchCounter, *numSubset, shardId, leaders, nodes, clientNode, clientPort)
+		}
+		batchCounter++
+		time.Sleep(1000 * time.Millisecond)
+	}
+
+	// Send a stop message to stop the nodes at the end
+	msg := proto_node.ConstructStopMessage()
+	peers := append(config.GetValidators(), leaders...)
+	p2p.BroadcastMessage(peers, msg)
+}
 
+func constructAndSendTransaction(subsetId, numSubset, shardId int, leaders []p2p.Peer, nodes []*node.Node, clientNode *node.Node, clientPort string) {
 	allCrossTxs := []*blockchain.Transaction{}
 	// Generate simulated transactions
-		for i, leader := range leaders {
-			txs, crossTxs := generateSimulatedTransactions(i, nodes)
+	leader := leaders[shardId]
+
+	txs, crossTxs := generateSimulatedTransactions(subsetId, numSubset, shardId, nodes)
 	allCrossTxs = append(allCrossTxs, crossTxs...)
 
 	log.Debug("[Generator] Sending single-shard txs ...", "leader", leader, "numTxs", len(txs), "numCrossTxs", len(crossTxs))
 	msg := proto_node.ConstructTransactionListMessage(txs)
 	p2p.SendMessage(leader, msg)
 	// Note cross shard txs are later sent in batch
-		}
 
 	if len(allCrossTxs) > 0 {
 		log.Debug("[Generator] Broadcasting cross-shard txs ...", "allCrossTxs", len(allCrossTxs))
@@ -343,12 +360,4 @@ func main() {
 			clientNode.Client.PendingCrossTxsMutex.Unlock()
 		}
 	}
-
-		time.Sleep(500 * time.Millisecond) // Send a batch of transactions periodically
-	}
-
-	// Send a stop message to stop the nodes at the end
-	msg := proto_node.ConstructStopMessage()
-	peers := append(config.GetValidators(), leaders...)
-	p2p.BroadcastMessage(peers, msg)
 }

consensus/consensus_leader.go

@@ -6,11 +6,12 @@ import (
 	"encoding/gob"
 	"encoding/hex"
 	"errors"
-	"net/http"
 	"net/url"
 	"strconv"
 	"time"
 
+	"github.com/simple-rules/harmony-benchmark/profiler"
+
 	"github.com/dedis/kyber"
 	"github.com/dedis/kyber/sign/schnorr"
 	"github.com/simple-rules/harmony-benchmark/blockchain"
@@ -404,7 +405,10 @@ func (consensus *Consensus) verifyResponse(commitments *map[uint16]kyber.Point,
 }
 
 func (consensus *Consensus) reportMetrics(block blockchain.Block) {
-	if !block.IsStateBlock() { // Skip state block stats
+	if block.IsStateBlock() { // Skip state block stats
+		return
+	}
+
 	endTime := time.Now()
 	timeElapsed := endTime.Sub(startTime)
 	numOfTxs := block.NumTransactions
@@ -418,14 +422,18 @@ func (consensus *Consensus) reportMetrics(block blockchain.Block) {
 		"consensus", consensus)
 
 	// Post metrics
-		URL := "http://localhost:3000/report"
+	profiler := profiler.GetProfiler()
+	if profiler.MetricsReportURL == "" {
+		return
+	}
+
 	txHashes := []string{}
 	for i := 1; i <= 3; i++ {
 		if len(block.TransactionIds)-i >= 0 {
 			txHashes = append(txHashes, hex.EncodeToString(block.TransactionIds[len(block.TransactionIds)-i][:]))
 		}
 	}
-		form := url.Values{
+	metrics := url.Values{
 		"key":             {consensus.pubKey.String()},
 		"tps":             {strconv.FormatFloat(tps, 'f', 2, 64)},
 		"txCount":         {strconv.Itoa(int(numOfTxs))},
@@ -434,11 +442,5 @@ func (consensus *Consensus) reportMetrics(block blockchain.Block) {
 		"latestTxHashes":  txHashes,
 		"blockLatency":    {strconv.Itoa(int(timeElapsed / time.Millisecond))},
 	}
-
-		body := bytes.NewBufferString(form.Encode())
-		rsp, err := http.Post(URL, "application/x-www-form-urlencoded", body)
-		if err == nil {
-			defer rsp.Body.Close()
-		}
-	}
+	profiler.LogMetrics(metrics)
 }

node/node.go

@@ -55,7 +55,8 @@ func (node *Node) addPendingTransactions(newTxs []*blockchain.Transaction) {
 // Note the pending transaction list will then contain the rest of the txs
 func (node *Node) getTransactionsForNewBlock(maxNumTxs int) ([]*blockchain.Transaction, []*blockchain.CrossShardTxAndProof) {
 	node.pendingTxMutex.Lock()
-	selected, unselected, crossShardTxs := node.UtxoPool.SelectTransactionsForNewBlock(node.pendingTransactions, maxNumTxs)
+	selected, unselected, invalid, crossShardTxs := node.UtxoPool.SelectTransactionsForNewBlock(node.pendingTransactions, maxNumTxs)
+	_ = invalid // invalid txs are discard
 	node.pendingTransactions = unselected
 	node.pendingTxMutex.Unlock()
 	return selected, crossShardTxs

node/node_handler.go

@@ -18,9 +18,9 @@ import (
 
 const (
 	// The max number of transaction per a block.
-	MaxNumberOfTransactionsPerBlock = 3000
+	MaxNumberOfTransactionsPerBlock = 10000
 	// The number of blocks allowed before generating state block
-	NumBlocksBeforeStateBlock = 10
+	NumBlocksBeforeStateBlock = 100
 )
 
 // NodeHandler handles a new incoming connection.

profiler/profiler.go

@@ -1,6 +1,11 @@
 package profiler
 
 import (
+	"bytes"
+	"net/http"
+	"net/url"
+	"os"
+	"sync"
 	"time"
 
 	"github.com/shirou/gopsutil/process"
@@ -8,15 +13,30 @@ import (
 )
 
 type Profiler struct {
+	// parameters
 	logger           log.Logger
-	PID     int32
-	ShardID string
+	pid              int32
+	shardID          string
+	MetricsReportURL string
+	// Internal
 	proc *process.Process
 }
 
-func NewProfiler(logger log.Logger, pid int, shardID string) *Profiler {
-	profiler := Profiler{logger, int32(pid), shardID, nil}
-	return &profiler
+var singleton *Profiler
+var once sync.Once
+
+func GetProfiler() *Profiler {
+	once.Do(func() {
+		singleton = &Profiler{}
+	})
+	return singleton
+}
+
+func (profiler *Profiler) Config(logger log.Logger, shardID string, metricsReportURL string) {
+	profiler.logger = logger
+	profiler.pid = int32(os.Getpid())
+	profiler.shardID = shardID
+	profiler.MetricsReportURL = metricsReportURL
 }
 
 func (profiler *Profiler) LogMemory() {
@@ -24,7 +44,7 @@ func (profiler *Profiler) LogMemory() {
 		// log mem usage
 		info, _ := profiler.proc.MemoryInfo()
 		memMap, _ := profiler.proc.MemoryMaps(false)
-		profiler.logger.Info("Mem Report", "info", info, "map", memMap, "shardID", profiler.ShardID)
+		profiler.logger.Info("Mem Report", "info", info, "map", memMap, "shardID", profiler.shardID)
 
 		time.Sleep(3 * time.Second)
 	}
@@ -35,14 +55,22 @@ func (profiler *Profiler) LogCPU() {
 		// log cpu usage
 		percent, _ := profiler.proc.CPUPercent()
 		times, _ := profiler.proc.Times()
-		profiler.logger.Info("CPU Report", "percent", percent, "times", times, "shardID", profiler.ShardID)
+		profiler.logger.Info("CPU Report", "percent", percent, "times", times, "shardID", profiler.shardID)
 
 		time.Sleep(3 * time.Second)
 	}
 }
 
+func (profiler *Profiler) LogMetrics(metrics url.Values) {
+	body := bytes.NewBufferString(metrics.Encode())
+	rsp, err := http.Post(profiler.MetricsReportURL, "application/x-www-form-urlencoded", body)
+	if err == nil {
+		defer rsp.Body.Close()
+	}
+}
+
 func (profiler *Profiler) Start() {
-	profiler.proc, _ = process.NewProcess(profiler.PID)
+	profiler.proc, _ = process.NewProcess(profiler.pid)
 	go profiler.LogCPU()
 	go profiler.LogMemory()
 }