Ricl update log

consensus/consensus.go

@@ -2,11 +2,12 @@
 package consensus // consensus
 
 import (
+	"fmt"
+	"harmony-benchmark/message"
 	"harmony-benchmark/p2p"
-	"regexp"
 	"log"
+	"regexp"
 	"strconv"
-	"harmony-benchmark/message"
 )
 
 // Consensus data containing all info related to one consensus process
@@ -40,7 +41,7 @@ type Consensus struct {
 
 	//// Network related fields
 	msgCategory byte
-	actionType byte
+	actionType  byte
 }
 
 // Consensus state enum for both leader and validator
@@ -119,10 +120,30 @@ func NewConsensus(ip, port string, peers []p2p.Peer, leader p2p.Peer) Consensus
 	return consensus
 }
 
-
 // Reset the state of the consensus
 func (consensus *Consensus) ResetState() {
 	consensus.state = READY
 	consensus.commits = make(map[string]string)
 	consensus.responses = make(map[string]string)
-}
+}
+
+// Returns ID of this consensus
+func (consensus *Consensus) GetIdentityString() string {
+	var duty string
+	if consensus.IsLeader {
+		duty = "LDR" // leader
+	} else {
+		duty = "SLV" // slave
+	}
+	return fmt.Sprintf("[%s, %s, %v]", duty, consensus.priKey, consensus.nodeId)
+}
+
+// Prints log with ID of this consensus
+func (consensus *Consensus) Logln(v ...interface{}) {
+	log.Printf("%s %s", consensus.GetIdentityString(), fmt.Sprintln(v...))
+}
+
+// Prints formatted log with ID of this consensus
+func (consensus *Consensus) Logf(format string, v ...interface{}) {
+	log.Printf("%s %s", consensus.GetIdentityString(), fmt.Sprintf(format, v...))
+}

consensus/consensus_leader.go

@@ -37,20 +37,20 @@ func (consensus *Consensus) ProcessMessageLeader(message []byte) {
 		log.Print(err)
 	}
 
-	log.Printf("[Leader] Received and processing message: %s\n", msgType)
+	consensus.Logf("Received and processing message: %s\n", msgType)
 	switch msgType {
 	case ANNOUNCE:
-		log.Printf("Unexpected message type: %s", msgType)
+		consensus.Logf("Unexpected message type: %s", msgType)
 	case COMMIT:
 		consensus.processCommitMessage(payload)
 	case CHALLENGE:
-		log.Printf("Unexpected message type: %s", msgType)
+		consensus.Logf("Unexpected message type: %s", msgType)
 	case RESPONSE:
 		consensus.processResponseMessage(payload)
 	case START_CONSENSUS:
 		consensus.processStartConsensusMessage(payload)
 	default:
-		log.Printf("Unexpected message type: %s", msgType)
+		consensus.Logf("Unexpected message type: %s", msgType)
 	}
 }
 
@@ -167,7 +167,7 @@ func (consensus *Consensus) processCommitMessage(payload []byte) {
 	shouldProcess := !ok && consensus.state == ANNOUNCE_DONE
 	if shouldProcess {
 		consensus.commits[validatorId] = validatorId
-		log.Printf("Number of commits received: %d", len(consensus.commits))
+		consensus.Logf("Number of commits received: %d", len(consensus.commits))
 	}
 	mutex.Unlock()
 
@@ -177,7 +177,7 @@ func (consensus *Consensus) processCommitMessage(payload []byte) {
 
 	mutex.Lock()
 	if len(consensus.commits) >= (2*len(consensus.validators))/3+1 {
-		log.Printf("Enough commits received with %d signatures", len(consensus.commits))
+		consensus.Logf("Enough commits received with %d signatures", len(consensus.commits))
 		if consensus.state == ANNOUNCE_DONE {
 			// Set state to CHALLENGE_DONE
 			consensus.state = CHALLENGE_DONE
@@ -272,7 +272,7 @@ func (consensus *Consensus) processResponseMessage(payload []byte) {
 	shouldProcess := !ok && consensus.state == CHALLENGE_DONE
 	if shouldProcess {
 		consensus.responses[validatorId] = validatorId
-		log.Printf("Number of responses received: %d", len(consensus.responses))
+		consensus.Logf("Number of responses received: %d", len(consensus.responses))
 	}
 	mutex.Unlock()
 
@@ -282,12 +282,12 @@ func (consensus *Consensus) processResponseMessage(payload []byte) {
 
 	mutex.Lock()
 	if len(consensus.responses) >= (2*len(consensus.validators))/3+1 {
-		log.Printf("Consensus reached with %d signatures.", len(consensus.responses))
+		consensus.Logf("Consensus reached with %d signatures.", len(consensus.responses))
 		if consensus.state == CHALLENGE_DONE {
 			// 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))
+			consensus.Logf("HOORAY!!! CONSENSUS REACHED AMONG %d NODES!!!\n", len(consensus.validators))
 			consensus.ResetState()
 			consensus.ReadySignal <- 1
 		}

consensus/consensus_validator.go

@@ -19,18 +19,18 @@ func (consensus *Consensus) ProcessMessageValidator(message []byte) {
 		log.Print(err)
 	}
 
-	log.Printf("[Validator] Received and processing message: %s\n", msgType)
+	consensus.Logf("Received and processing message: %s\n", msgType)
 	switch msgType {
 	case ANNOUNCE:
 		consensus.processAnnounceMessage(payload)
 	case COMMIT:
-		log.Printf("Unexpected message type: %s", msgType)
+		consensus.Logf("Unexpected message type: %s", msgType)
 	case CHALLENGE:
 		consensus.processChallengeMessage(payload)
 	case RESPONSE:
-		log.Printf("Unexpected message type: %s", msgType)
+		consensus.Logf("Unexpected message type: %s", msgType)
 	default:
-		log.Printf("Unexpected message type: %s", msgType)
+		consensus.Logf("Unexpected message type: %s", msgType)
 	}
 }
 

node/node.go

@@ -52,41 +52,25 @@ func (node *Node) NodeHandler(conn net.Conn) {
 
 	consensus := node.consensus
 	if err != nil {
-		if consensus.IsLeader {
-			log.Printf("[Leader] Read p2p data failed:%s", err)
-		} else {
-			log.Printf("[Slave] Read p2p data failed:%s", err)
-		}
+		node.Logf("Read p2p data failed:%s", err)
 		return
 	}
 
 	msgCategory, err := message.GetMessageCategory(content)
 	if err != nil {
-		if consensus.IsLeader {
-			log.Printf("[Leader] Read node type failed:%s", err)
-		} else {
-			log.Printf("[Slave] Read node type failed:%s", err)
-		}
+		node.Logf("Read node type failed:%s", err)
 		return
 	}
 
 	msgType, err := message.GetMessageType(content)
 	if err != nil {
-		if consensus.IsLeader {
-			log.Printf("[Leader] Read action type failed:%s", err)
-		} else {
-			log.Printf("[Slave] Read action type failed:%s", err)
-		}
+		node.Logf("Read action type failed:%s", err)
 		return
 	}
 
 	msgPayload, err := message.GetMessagePayload(content)
 	if err != nil {
-		if consensus.IsLeader {
-			log.Printf("[Leader] Read message payload failed:%s", err)
-		} else {
-			log.Printf("[Slave] Read message payload failed:%s", err)
-		}
+		node.Logf("Read message payload failed:%s", err)
 		return
 	}
 
@@ -110,14 +94,14 @@ func (node *Node) NodeHandler(conn net.Conn) {
 			txList := new([]blockchain.Transaction)
 			err := txDecoder.Decode(&txList)
 			if err != nil {
-				log.Println("Failed deserializing transaction list")
+				node.Logln("Failed deserializing transaction list")
 			}
 			node.pendingTransactions = append(node.pendingTransactions, *txList...)
-			log.Println(len(node.pendingTransactions))
+			node.Logln(len(node.pendingTransactions))
 		case message.CONTROL:
 			controlType := msgPayload[0]
 			if ControlMessageType(controlType) == STOP {
-				log.Println("Stopping Node")
+				node.Logln("Stopping Node")
 				os.Exit(0)
 			}
 
@@ -132,7 +116,7 @@ func (node *Node) WaitForConsensusReady(readySignal chan int) {
 		newBlock := new(blockchain.Block)
 		for {
 			if len(node.pendingTransactions) >= 10 {
-				log.Println("Creating new block")
+				node.Logln("Creating new block")
 				// TODO (Minh): package actual transactions
 				// For now, just take out 10 transactions
 				var txList []*blockchain.Transaction
@@ -156,3 +140,18 @@ func NewNode(consensus *consensus.Consensus) Node {
 	node.BlockChannel = make(chan blockchain.Block)
 	return node
 }
+
+// Returns the identity string of this node
+func (node *Node) GetIdentityString() string {
+	return node.consensus.GetIdentityString()
+}
+
+// Prints log with ID of this node
+func (node *Node) Logln(v ...interface{}) {
+	node.consensus.Logln(v...)
+}
+
+// Prints formatted log with ID of this node
+func (node *Node) Logf(format string, v ...interface{}) {
+	node.consensus.Logf(format, v...)
+}