Add challenge broadcast; put consensus signature in map; refactor consensus init into consensus package

benchmark_node.go

@@ -128,24 +128,6 @@ func NodeHandler(conn net.Conn, consensus *consensus.Consensus) {
 	//relayToPorts(receivedMessage, conn)
 }
 
-func initConsensus(ip, port, ipfile string) consensus.Consensus {
-	// The first Ip, port passed will be leader.
-	consensus := consensus.Consensus{}
-	peer := p2p.Peer{Port: port, Ip: ip}
-	Peers := getPeers(ip, port, ipfile)
-	leaderPeer := getLeader(ipfile)
-	if leaderPeer == peer {
-		consensus.IsLeader = true
-	} else {
-		consensus.IsLeader = false
-	}
-	consensus.Leader = leaderPeer
-	consensus.Validators = Peers
-
-	consensus.PriKey = ip + ":" + port // use ip:port as unique key for now
-	return consensus
-}
-
 func getLeader(iplist string) p2p.Peer {
 	file, _ := os.Open(iplist)
 	fscanner := bufio.NewScanner(file)
@@ -183,15 +165,16 @@ func main() {
 	ipfile := flag.String("ipfile", "iplist.txt", "file containing all ip addresses")
 	flag.Parse()
 	fmt.Println()
-	consensus := initConsensus(*ip, *port, *ipfile)
+
+	consensusObj := consensus.InitConsensus(*ip, *port, getPeers(*ip, *port, *ipfile), getLeader(*ipfile))
 	var nodeStatus string
-	if consensus.IsLeader {
+	if consensusObj.IsLeader {
 		nodeStatus = "leader"
 	} else {
 		nodeStatus = "validator"
 	}
-	fmt.Println(consensus)
+	fmt.Println(consensusObj)
 	fmt.Printf("This node is a %s node with ip: %s and port: %s\n", nodeStatus, *ip, *port)
 	fmt.Println()
-	startServer(*port, NodeHandler, &consensus)
+	startServer(*port, NodeHandler, &consensusObj)
 }

consensus/consensus.go

@@ -9,7 +9,7 @@ import (
 type Consensus struct {
 	State ConsensusState
 	// Signatures collected from validators
-	Signatures []string
+	Signatures map[string]string
 	// Actual block data to reach consensus on
 	Data string
 	// List of validators
@@ -53,3 +53,22 @@ func (state ConsensusState) String() string {
 	}
 	return names[state]
 }
+
+func InitConsensus(ip, port string, peers []p2p.Peer, leader p2p.Peer) Consensus {
+	// The first Ip, port passed will be leader.
+	consensus := Consensus{}
+	peer := p2p.Peer{Port: port, Ip: ip}
+	Peers := peers
+	leaderPeer := leader
+	if leaderPeer == peer {
+		consensus.IsLeader = true
+	} else {
+		consensus.IsLeader = false
+	}
+	consensus.Signatures = make(map[string]string)
+	consensus.Leader = leaderPeer
+	consensus.Validators = Peers
+
+	consensus.PriKey = ip + ":" + port // use ip:port as unique key for now
+	return consensus
+}

consensus/consensus_leader.go

@@ -55,16 +55,26 @@ func (consensus *Consensus) startConsensus(msg string) {
 
 func (consensus *Consensus) processCommitMessage(msg string) {
 	// verify and aggregate all the signatures
-	mutex.Lock()
+	if _, ok := consensus.Signatures[msg]; !ok {
-	consensus.Signatures = append(consensus.Signatures, msg)
+		mutex.Lock()
+		consensus.Signatures[msg] = msg
+		mutex.Unlock()
+	}
 
-	// Broadcast challenge
-	// Set state to CHALLENGE_DONE
-	consensus.State = CHALLENGE_DONE
-	mutex.Unlock()
 
 	log.Printf("Number of signatures received: %d", len(consensus.Signatures))
-	if len(consensus.Signatures) >= (2 * len(consensus.Validators)) / 3 + 1 {
+	if consensus.State != CHALLENGE_DONE && len(consensus.Signatures) >= (2 * len(consensus.Validators)) / 3 + 1 {
+
+		mutex.Lock()
+		if consensus.State == ANNOUNCE_DONE {
+			// Set state to CHALLENGE_DONE
+			consensus.State = CHALLENGE_DONE
+		}
+		mutex.Unlock()
+		// Broadcast challenge
+		msgToSend := ConstructConsensusMessage(CHALLENGE, []byte("challenge"))
+		p2p.BroadcastMessage(consensus.Validators, msgToSend)
+
 		log.Printf("Consensus reached with %d signatures: %s", len(consensus.Signatures), consensus.Signatures)
 	}