WIP: adding message to stop running consensus

aws-code/transaction_generator.go

@@ -1,12 +1,12 @@
 package main
 
 import (
-	"harmony-benchmark/blockchain"
-	"math/rand"
-	"time"
 	"flag"
+	"harmony-benchmark/blockchain"
 	"harmony-benchmark/node"
 	"harmony-benchmark/p2p"
+	"math/rand"
+	"time"
 )
 
 func newRandTransaction() blockchain.Transaction {
@@ -22,9 +22,14 @@ func main() {
 
 	ip := flag.String("ip", "127.0.0.1", "IP of the leader")
 	port := flag.String("port", "9000", "port of the leader.")
-
+	start := time.Now()
+	totalTime = 30000
 	txs := make([]blockchain.Transaction, 10)
 	for true {
+		t = time.Now()
+		if t.Sub(start) >= totalTime {
+			break
+		}
 		for i := range txs {
 			txs[i] = newRandTransaction()
 
@@ -33,4 +38,6 @@ func main() {
 		p2p.SendMessage(p2p.Peer{*ip, *port, "n/a"}, msg)
 		time.Sleep(1 * time.Second) // 10 transactions per second
 	}
+	msg := node.ConstructTransactionListMessage(txs)
+	p2p.SendMessage(p2p.Peer{*ip, *port, "n/a"}, msg)
 }

consensus/consensus_leader.go

@@ -4,11 +4,11 @@ import (
 	"log"
 	"sync"
 
-	"harmony-benchmark/p2p"
 	"bytes"
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"harmony-benchmark/p2p"
 )
 
 var mutex = &sync.Mutex{}
@@ -38,6 +38,8 @@ func (consensus *Consensus) ProcessMessageLeader(message []byte) {
 		consensus.processResponseMessage(payload)
 	case START_CONSENSUS:
 		consensus.processStartConsensusMessage(msg)
+	case STOP_CONSENSUS:
+		consensus.stopConsensusMessage(msg)
 	default:
 		log.Println("Unexpected message type: %s", msgType)
 	}
@@ -119,23 +121,23 @@ func (consensus *Consensus) processCommitMessage(payload []byte) {
 	//#### Read payload data
 	offset := 0
 	// 4 byte consensus id
-	consensusId := binary.BigEndian.Uint32(payload[offset:offset+4])
+	consensusId := binary.BigEndian.Uint32(payload[offset : offset+4])
 	offset += 4
 
 	// 32 byte block hash
-	blockHash := payload[offset:offset+32]
+	blockHash := payload[offset : offset+32]
 	offset += 32
 
 	// 2 byte validator id
-	validatorId := string(payload[offset:offset+2])
+	validatorId := string(payload[offset : offset+2])
 	offset += 2
 
 	// 33 byte commit
-	commit := payload[offset:offset+33]
+	commit := payload[offset : offset+33]
 	offset += 33
 
 	// 64 byte of signature on previous data
-	signature := payload[offset:offset+64]
+	signature := payload[offset : offset+64]
 	offset += 64
 	//#### END: Read payload data
 
@@ -225,23 +227,23 @@ func (consensus *Consensus) processResponseMessage(payload []byte) {
 	//#### Read payload data
 	offset := 0
 	// 4 byte consensus id
-	consensusId := binary.BigEndian.Uint32(payload[offset:offset+4])
+	consensusId := binary.BigEndian.Uint32(payload[offset : offset+4])
 	offset += 4
 
 	// 32 byte block hash
-	blockHash := payload[offset:offset+32]
+	blockHash := payload[offset : offset+32]
 	offset += 32
 
 	// 2 byte validator id
-	validatorId := string(payload[offset:offset+2])
+	validatorId := string(payload[offset : offset+2])
 	offset += 2
 
 	// 32 byte response
-	response := payload[offset:offset+32]
+	response := payload[offset : offset+32]
 	offset += 32
 
 	// 64 byte of signature on previous data
-	signature := payload[offset:offset+64]
+	signature := payload[offset : offset+64]
 	offset += 64
 	//#### END: Read payload data
 

consensus/message.go

@@ -78,6 +78,7 @@ const (
 	CHALLENGE
 	RESPONSE
 	START_CONSENSUS
+	STOP_CONSENSUS
 )
 
 // Returns string name for the MessageType enum
@@ -87,9 +88,11 @@ func (msgType MessageType) String() string {
 		"COMMIT",
 		"CHALLENGE",
 		"RESPONSE",
-		"START_CONSENSUS"}
+		"START_CONSENSUS"
+		"STOP_CONSENSUS"
+	}
 
-	if msgType < ANNOUNCE || msgType > START_CONSENSUS {
+	if msgType < ANNOUNCE || msgType > STOP_CONSENSUS {
 		return "Unknown"
 	}
 	return names[msgType]