merging conflicts

aws-code/transaction_generator.go

@@ -46,6 +46,7 @@ func main() {
 	start := time.Now()
 	totalTime := 60.0
 	txs := make([]blockchain.Transaction, 10)
+	txCount := 0
 	for true {
 		t := time.Now()
 		if t.Sub(start).Seconds() >= totalTime {

benchmark_main.go

@@ -61,5 +61,19 @@ func main() {
 	log.Println("======================================")
 
 	node := node.NewNode(&consensus)
+
+	if consensus.IsLeader {
+		// Let consensus run
+		go func() {
+			log.Println("Waiting for block")
+			consensus.WaitForNewBlock(node.BlockChannel)
+		}()
+		// Node waiting for consensus readiness to create new block
+		go func() {
+			log.Println("Waiting for consensus ready")
+			node.WaitForConsensusReady(consensus.ReadySignal)
+		}()
+	}
+
 	node.StartServer(*port)
 }

blockchain/block.go

@@ -62,7 +62,7 @@ func (b *Block) HashTransactions() []byte {
 	return txHash[:]
 }
 
-// NewBlock creates and returns Block.
+// NewBlock creates and returns a neew block.
 func NewBlock(transactions []*Transaction, prevBlockHash []byte) *Block {
 	block := &Block{time.Now().Unix(), transactions, prevBlockHash, []byte{}}
 	block.Hash = block.HashTransactions()

blockchain/blockchain.go

@@ -1,7 +1,9 @@
 package blockchain
 
 import (
+	"bytes"
 	"encoding/hex"
+	"fmt"
 )
 
 // Blockchain keeps a sequence of Blocks
@@ -134,13 +136,28 @@ func (bc *Blockchain) NewUTXOTransaction(from, to string, amount int) *Transacti
 func (bc *Blockchain) AddNewTransferAmount(from, to string, amount int) *Blockchain {
 	tx := bc.NewUTXOTransaction(from, to, amount)
 	if tx != nil {
-		newBlock := NewBlock([]*Transaction{tx}, bc.blocks[len(bc.blocks)-1].PrevBlockHash)
+		newBlock := NewBlock([]*Transaction{tx}, bc.blocks[len(bc.blocks)-1].Hash)
 		bc.blocks = append(bc.blocks, newBlock)
 		return bc
 	}
 	return nil
 }
 
+// VerifyNewBlock verifies if the new coming block is valid for the current blockchain.
+func (bc *Blockchain) VerifyNewBlock(block *Block) bool {
+	length := len(bc.blocks)
+	if bytes.Compare(block.PrevBlockHash, bc.blocks[length-1].Hash) != 0 {
+		fmt.Println("MINh1")
+		return false
+	}
+	if block.Timestamp < bc.blocks[length-1].Timestamp {
+		fmt.Println("MINh2")
+		return false
+	}
+	// TODO(minhdoan): Check Transactions parts
+	return true
+}
+
 // CreateBlockchain creates a new blockchain DB
 func CreateBlockchain(address string) *Blockchain {
 	// TODO: We assume we have not created any blockchain before.

blockchain/blockchain_test.go

@@ -61,3 +61,19 @@ func TestAddNewTransferAmount(t *testing.T) {
 		t.Error("minh should not have enough fun to make the transfer")
 	}
 }
+
+func TestVerifyNewBlock(t *testing.T) {
+	bc := CreateBlockchain("minh")
+	bc = bc.AddNewTransferAmount("minh", "alok", 3)
+	bc = bc.AddNewTransferAmount("minh", "rj", 100)
+
+	tx := bc.NewUTXOTransaction("minh", "mark", 10)
+	if tx == nil {
+		t.Error("failed to create a new transaction.")
+	}
+	newBlock := NewBlock([]*Transaction{tx}, bc.blocks[len(bc.blocks)-1].Hash)
+
+	if !bc.VerifyNewBlock(newBlock) {
+		t.Error("failed to add a new valid block.")
+	}
+}

consensus/consensus.go

@@ -35,6 +35,9 @@ type Consensus struct {
 	// BlockHeader to run consensus on
 	blockHeader []byte
 
+	// Signal channel for starting a new consensus process
+	ReadySignal chan int
+
 	//// Network related fields
 	msgCategory byte
 	actionType byte
@@ -103,6 +106,14 @@ func NewConsensus(ip, port string, peers []p2p.Peer, leader p2p.Peer) Consensus
 	value, err := strconv.Atoi(socketId)
 	consensus.nodeId = uint16(value)
 
+	if consensus.IsLeader {
+		consensus.ReadySignal = make(chan int)
+		// send a signal to indicate it's ready to run consensus
+		go func() {
+			consensus.ReadySignal <- 1
+		}()
+	}
+
 	consensus.msgCategory = byte(message.COMMITTEE)
 	consensus.actionType = byte(message.CONSENSUS)
 	return consensus

consensus/consensus_leader.go

@@ -8,11 +8,22 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"harmony-benchmark/blockchain"
 	"harmony-benchmark/p2p"
 )
 
 var mutex = &sync.Mutex{}
 
+func (consensus *Consensus) WaitForNewBlock(blockChannel chan blockchain.Block) {
+	for { // keep waiting for new blocks
+		newBlock := <-blockChannel
+		// TODO: think about potential race condition
+		if consensus.state == READY {
+			consensus.startConsensus(&newBlock)
+		}
+	}
+}
+
 // Leader's consensus message dispatcher
 func (consensus *Consensus) ProcessMessageLeader(message []byte) {
 	msgType, err := GetConsensusMessageType(message)
@@ -25,7 +36,6 @@ func (consensus *Consensus) ProcessMessageLeader(message []byte) {
 		log.Print(err)
 	}
 
-	msg := string(payload)
 	log.Printf("[Leader] Received and processing message: %s\n", msgType)
 	switch msgType {
 	case ANNOUNCE:
@@ -37,18 +47,22 @@ func (consensus *Consensus) ProcessMessageLeader(message []byte) {
 	case RESPONSE:
 		consensus.processResponseMessage(payload)
 	case START_CONSENSUS:
-		consensus.processStartConsensusMessage(msg)
+		consensus.processStartConsensusMessage(payload)
 	default:
 		log.Println("Unexpected message type: %s", msgType)
 	}
 }
 
 // Handler for message which triggers consensus process
-func (consensus *Consensus) processStartConsensusMessage(msg string) {
+func (consensus *Consensus) processStartConsensusMessage(payload []byte) {
+	consensus.startConsensus(blockchain.NewGenesisBlock(blockchain.NewCoinbaseTX("x", "y")))
+}
+
+func (consensus *Consensus) startConsensus(newBlock *blockchain.Block) {
 	// prepare message and broadcast to validators
 	// Construct new block
-	newBlock := constructNewBlock()
-	consensus.blockHash = getBlockHash(newBlock)
+	//newBlock := constructNewBlock()
+	consensus.blockHash = newBlock.Hash
 
 	msgToSend, err := consensus.constructAnnounceMessage()
 	if err != nil {
@@ -273,6 +287,7 @@ func (consensus *Consensus) processResponseMessage(payload []byte) {
 			// TODO: do followups on the consensus
 			log.Printf("HOORAY!!! CONSENSUS REACHED AMONG %d NODES!!!\n", len(consensus.validators))
 			consensus.ResetState()
+			consensus.ReadySignal <- 1
 		}
 		// TODO: composes new block and broadcast the new block to validators
 	}

local_iplist.txt

@@ -1,4 +1,3 @@
-127.0.0.1 9000 leader
 127.0.0.1 9001 validator
 127.0.0.1 9002 validator
 127.0.0.1 9003 validator
@@ -99,3 +98,4 @@
 127.0.0.1 9098 validator
 127.0.0.1 9099 validator
 127.0.0.1 9100 validator
+127.0.0.1 9000 leader

message/message.go

@@ -10,13 +10,14 @@ Node will process the content of the p2p message
 
 ----  content start -----
 1 byte            - message category
-                    0x00: consensus
-                    0x01: normal...
-1 byte            - action type
-                    - consensus node
+                    0x00: COMMITTEE
+                    0x01: NODE...
+1 byte            - message type
+                    - for COMMITTEE category
                       0x00: consensus
-				    - normal node
-                      0x00: transaction
+                      0x01: sharding ...
+				    - for NODE category
+                      0x00: transaction ...
 n - 2 bytes       - actual message payload
 ----   content end  -----
 

node/node.go

@@ -10,16 +10,19 @@ import (
 	"log"
 	"net"
 	"os"
+	"time"
 )
 
 // A node represents a program (machine) participating in the network
 type Node struct {
 	consensus           *consensus.Consensus
+	consensus           *consensus.Consensus
+	BlockChannel        chan blockchain.Block
 	pendingTransactions []blockchain.Transaction
 }
 
 // Start a server and process the request by a handler.
-func (node Node) StartServer(port string) {
+func (node *Node) StartServer(port string) {
 	listenOnPort(port, node.NodeHandler)
 }
 
@@ -46,7 +49,7 @@ func (node *Node) NodeHandler(conn net.Conn) {
 	defer conn.Close()
 
 	// Read p2p message payload
-	payload, err := p2p.ReadMessageContent(conn)
+	content, err := p2p.ReadMessageContent(conn)
 
 	consensus := node.consensus
 	if err != nil {
@@ -58,7 +61,7 @@ func (node *Node) NodeHandler(conn net.Conn) {
 		return
 	}
 
-	msgCategory, err := message.GetMessageCategory(payload)
+	msgCategory, err := message.GetMessageCategory(content)
 	if err != nil {
 		if consensus.IsLeader {
 			log.Printf("[Leader] Read node type failed:%s", err)
@@ -68,7 +71,7 @@ func (node *Node) NodeHandler(conn net.Conn) {
 		return
 	}
 
-	msgType, err := message.GetMessageType(payload)
+	msgType, err := message.GetMessageType(content)
 	if err != nil {
 		if consensus.IsLeader {
 			log.Printf("[Leader] Read action type failed:%s", err)
@@ -78,7 +81,7 @@ func (node *Node) NodeHandler(conn net.Conn) {
 		return
 	}
 
-	msgPayload, err := message.GetMessagePayload(payload)
+	msgPayload, err := message.GetMessagePayload(content)
 	if err != nil {
 		if consensus.IsLeader {
 			log.Printf("[Leader] Read message payload failed:%s", err)
@@ -123,9 +126,34 @@ func (node *Node) NodeHandler(conn net.Conn) {
 	}
 }
 
+func (node *Node) WaitForConsensusReady(readySignal chan int) {
+	for { // keep waiting for consensus ready
+		<-readySignal
+		// create a new block
+		newBlock := new(blockchain.Block)
+		for {
+			if len(node.pendingTransactions) >= 10 {
+				log.Println("Creating new block")
+				// TODO (Minh): package actual transactions
+				// For now, just take out 10 transactions
+				var txList []*blockchain.Transaction
+				for _, tx := range node.pendingTransactions[0:10] {
+					txList = append(txList, &tx)
+				}
+				node.pendingTransactions = node.pendingTransactions[10:]
+				newBlock = blockchain.NewBlock(txList, []byte{})
+				break
+			}
+			time.Sleep(1 * time.Second) // Periodically check whether we have enough transactions to package into block.
+		}
+		node.BlockChannel <- *newBlock
+	}
+}
+
 // Create a new Node
 func NewNode(consensus *consensus.Consensus) Node {
 	node := Node{}
 	node.consensus = consensus
+	node.BlockChannel = make(chan blockchain.Block)
 	return node
 }