diff --git a/aws-code/transaction_generator.go b/aws-code/transaction_generator.go index 2cb8de967..8cdbc38fa 100644 --- a/aws-code/transaction_generator.go +++ b/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 { diff --git a/benchmark_main.go b/benchmark_main.go index a1cfa9474..ba30661f0 100644 --- a/benchmark_main.go +++ b/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) } diff --git a/blockchain/block.go b/blockchain/block.go index d3bfca4e5..2eb64fc31 100644 --- a/blockchain/block.go +++ b/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() diff --git a/blockchain/blockchain.go b/blockchain/blockchain.go index cec050860..edcab3389 100644 --- a/blockchain/blockchain.go +++ b/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. diff --git a/blockchain/blockchain_test.go b/blockchain/blockchain_test.go index 5db10a02e..0fb2003c9 100644 --- a/blockchain/blockchain_test.go +++ b/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.") + } +} diff --git a/consensus/consensus.go b/consensus/consensus.go index 3e508269b..a3ee43197 100644 --- a/consensus/consensus.go +++ b/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 diff --git a/consensus/consensus_leader.go b/consensus/consensus_leader.go index fc3c24124..155e20e8a 100644 --- a/consensus/consensus_leader.go +++ b/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 } diff --git a/local_iplist.txt b/local_iplist.txt index 47de6810a..9f5d026fe 100644 --- a/local_iplist.txt +++ b/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 diff --git a/message/message.go b/message/message.go index 6e2be292a..c951379c3 100644 --- a/message/message.go +++ b/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 ----- diff --git a/node/node.go b/node/node.go index 4f68608a4..116faa792 100644 --- a/node/node.go +++ b/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 }