Add sharding config with 2 shard; update tx generator accordingly

aws-code/transaction_generator.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"strings"
 	"time"
+	"log"
 )
 
 func newRandTransaction() blockchain.Transaction {
@@ -22,14 +23,14 @@ func newRandTransaction() blockchain.Transaction {
 	return tx
 }
 
-func getPeers(Ip, Port, config string) []p2p.Peer {
+func getValidators(config string) []p2p.Peer {
 	file, _ := os.Open(config)
 	fscanner := bufio.NewScanner(file)
 	var peerList []p2p.Peer
 	for fscanner.Scan() {
 		p := strings.Split(fscanner.Text(), " ")
 		ip, port, status := p[0], p[1], p[2]
-		if status == "leader" || ip == Ip && port == Port {
+		if status == "leader" {
 			continue
 		}
 		peer := p2p.Peer{Port: port, Ip: ip}
@@ -37,15 +38,39 @@ func getPeers(Ip, Port, config string) []p2p.Peer {
 	}
 	return peerList
 }
-func main() {
 
-	ip := flag.String("ip", "127.0.0.1", "IP of the leader")
+func getLeaders(config *[][]string) []p2p.Peer {
-	port := flag.String("port", "9000", "port of the leader.")
+	var peerList []p2p.Peer
+	for _, node := range *config {
+		ip, port, status := node[0], node[1], node[2]
+		if status == "leader" {
+			peerList = append(peerList, p2p.Peer{Ip: ip, Port: port})
+		}
+	}
+	return peerList
+}
+
+func readConfigFile(configFile string) [][]string{
+	file, _ := os.Open(configFile)
+	fscanner := bufio.NewScanner(file)
+
+	result := [][]string{}
+	for fscanner.Scan() {
+		p := strings.Split(fscanner.Text(), " ")
+		result = append(result, p)
+	}
+	return result
+}
+
+func main() {
 	configFile := flag.String("config_file", "local_config.txt", "file containing all ip addresses and config")
-	//getLeader to get ip,port and get totaltime I want to run
+	flag.Parse()
+	config := readConfigFile(*configFile)
+
 	start := time.Now()
 	totalTime := 60.0
 	txs := make([]blockchain.Transaction, 10)
+	leaders := getLeaders(&config)
 	for true {
 		t := time.Now()
 		if t.Sub(start).Seconds() >= totalTime {
@@ -57,13 +82,11 @@ func main() {
 
 		}
 		msg := node.ConstructTransactionListMessage(txs)
-		p2p.SendMessage(p2p.Peer{*ip, *port, "n/a"}, msg)
+		log.Printf("[Generator] Sending txs to %d leader[s]\n", len(leaders))
+		p2p.BroadcastMessage(leaders, msg)
 		time.Sleep(1 * time.Second) // 10 transactions per second
 	}
 	msg := node.ConstructStopMessage()
-	var leaderPeer p2p.Peer
+	peers := append(getValidators(*configFile), leaders...)
-	leaderPeer.Ip = *ip
-	leaderPeer.Port = *port
-	peers := append(getPeers(*ip, *port, *configFile), leaderPeer)
 	p2p.BroadcastMessage(peers, msg)
 }

benchmark_main.go

@@ -36,8 +36,8 @@ func getLeader(myShardId string, config *[][]string) p2p.Peer {
 func getPeers(myIp, myPort, myShardId string, config *[][]string) []p2p.Peer {
 	var peerList []p2p.Peer
 	for _, node := range *config {
-		ip, port, status := node[0], node[1], node[2]
+		ip, port, status, shardId := node[0], node[1], node[2], node[3]
-		if status == "leader" || ip == myIp && port == myPort {
+		if status == "leader" || ip == myIp && port == myPort || myShardId != shardId {
 			continue
 		}
 		peer := p2p.Peer{Port: port, Ip: ip}
@@ -79,7 +79,7 @@ func main() {
 	}
 
 	log.Println("======================================")
-	log.Printf("This node is a %s node listening on ip: %s and port: %s\n", nodeStatus, *ip, *port)
+	log.Printf("This node is a %s node in shard %s listening on ip: %s and port: %s\n", nodeStatus, shardId, *ip, *port)
 	log.Println("======================================")
 
 	node := node.NewNode(&consensus)

consensus/consensus_leader.go

@@ -37,7 +37,7 @@ func (consensus *Consensus) ProcessMessageLeader(message []byte) {
 		log.Print(err)
 	}
 
-	log.Printf("[Leader] Received and processing message: %s\n", msgType)
+	log.Printf("[Leader-%d] Received and processing message: %s\n", consensus.ShardId, msgType)
 	switch msgType {
 	case ANNOUNCE:
 		log.Printf("Unexpected message type: %s", msgType)
@@ -287,7 +287,7 @@ func (consensus *Consensus) processResponseMessage(payload []byte) {
 			// Set state to FINISHED
 			consensus.state = FINISHED
 			// TODO: do followups on the consensus
-			log.Printf("HOORAY!!! CONSENSUS REACHED AMONG %d NODES!!!\n", len(consensus.validators))
+			log.Printf("[Shard %d] HOORAY!!! CONSENSUS REACHED AMONG %d NODES WITH CONSENSUS ID %d!!!\n", consensus.ShardId, len(consensus.validators), consensus.consensusId)
 			consensus.ResetState()
 			consensus.consensusId++
 			consensus.ReadySignal <- 1

local_config_shards.txt

@@ -0,0 +1,22 @@
+127.0.0.1 9010 validator 0
+127.0.0.1 9011 validator 0
+127.0.0.1 9012 validator 0
+127.0.0.1 9013 validator 0
+127.0.0.1 9014 validator 0
+127.0.0.1 9015 validator 0
+127.0.0.1 9016 validator 0
+127.0.0.1 9017 validator 0
+127.0.0.1 9018 validator 0
+127.0.0.1 9019 validator 0
+127.0.0.1 9020 validator 1
+127.0.0.1 9021 validator 1
+127.0.0.1 9022 validator 1
+127.0.0.1 9023 validator 1
+127.0.0.1 9024 validator 1
+127.0.0.1 9025 validator 1
+127.0.0.1 9026 validator 1
+127.0.0.1 9027 validator 1
+127.0.0.1 9028 validator 1
+127.0.0.1 9029 validator 1
+127.0.0.1 9000 leader 0
+127.0.0.1 9001 leader 1