Merge pull request #89 from harmony-one/lc4pr

HAR-5: fix nonstop ping message and concurrent map access issue

deploy.sh

@@ -7,26 +7,44 @@ function usage {
    local ME=$(basename $0)
 
    cat<<EOU
-USAGE: $ME config_file_name
+USAGE: $ME [OPTIONS] config_file_name
+
+   -h             print this help message
+   -p             use peer discovery (default: $PEER)
+   -d             enable db support (default: $DB)
+   -t             toggle txgen (default: $TXGEN)
 
 This script will build all the binaries and start benchmark and txgen based on the configuration file.
 
 EXAMPLES:
 
    $ME local_config.txt
-
+   $ME -p local_config.txt
 
 EOU
    exit 0
 }
 
+PEER=
+DB=
+TXGEN=true
+
+while getopts "hpdt" option; do
+   case $option in
+      h) usage ;;
+      p) PEER='-peer_discovery' ;;
+      d) DB='-db_supported' ;;
+      t) TXGEN=false ;;
+   esac
+done
+
+shift $((OPTIND-1))
+
 config=$1
 if [ -z "$config" ]; then
    usage
 fi
 
-db_supported=$2
-
 # Kill nodes if any
 ./kill_node.sh
 
@@ -49,16 +67,10 @@ while IFS='' read -r line || [[ -n "$line" ]]; do
   IFS=' ' read ip port mode shardID <<< $line
 	#echo $ip $port $mode
   if [ "$mode" != "client" ]; then
-    if [ -z "$db_supported" ]; then
-      ./bin/benchmark -ip $ip -port $port -config_file $config -log_folder $log_folder&
-    else
-      ./bin/benchmark -ip $ip -port $port -config_file $config -log_folder $log_folder  -db_supported&
-    fi
+    ./bin/benchmark -ip $ip -port $port -config_file $config -log_folder $log_folder $DB $PEER &
   fi
 done < $config
 
-txgen_disabled=$3
-# Generate transactions
-if [ -z "$txgen_disabled" ]; then
+if [ "$TXGEN" == "true" ]; then
   ./bin/txgen -config_file $config -log_folder $log_folder
 fi

go_executable_build.sh

@@ -61,6 +61,9 @@ function build_only
       if [ "$(uname -s)" == "Linux" ]; then
          $BINDIR/$bin -version
       fi
+      if [ "$(uname -s)" == "Darwin" -a "$GOOS" == "darwin" ]; then
+         $BINDIR/$bin -version
+      fi
    done
 
    $MD5 $BINDIR/* > $BINDIR/md5sum.txt 2> /dev/null

node/node.go

@@ -64,10 +64,10 @@ type Node struct {
 	SelfPeer               p2p.Peer // TODO(minhdoan): it could be duplicated with Self below whose is Alok work.
 	IDCPeer                p2p.Peer
 
-	SyncNode  bool                 // TODO(minhdoan): Remove it later.
-	chain     *core.BlockChain     // Account Model
-	Neighbors map[string]*p2p.Peer // All the neighbor nodes, key is the sha256 of Peer IP/Port
-	State     NodeState            // State of the Node
+	SyncNode  bool             // TODO(minhdoan): Remove it later.
+	chain     *core.BlockChain // Account Model
+	Neighbors sync.Map         // All the neighbor nodes, key is the sha256 of Peer IP/Port, value is the p2p.Peer
+	State     NodeState        // State of the Node
 
 	// Account Model
 	Chain               *core.BlockChain
@@ -245,7 +245,6 @@ func New(consensus *bft.Consensus, db *hdb.LDBDatabase) *Node {
 	}
 	// Logger
 	node.log = log.New()
-	node.Neighbors = make(map[string]*p2p.Peer)
 	node.State = INIT
 
 	return &node
@@ -256,15 +255,14 @@ func (node *Node) AddPeers(peers []p2p.Peer) int {
 	count := 0
 	for _, p := range peers {
 		key := fmt.Sprintf("%v", p.PubKey)
-		_, ok := node.Neighbors[key]
+		_, ok := node.Neighbors.Load(key)
 		if !ok {
 			np := new(p2p.Peer)
 			copier.Copy(np, &p)
-			node.Neighbors[key] = np
+			node.Neighbors.Store(key, *np)
 			count++
 		}
 	}
-	node.log.Info("Added", "# of peers", count)
 
 	if count > 0 {
 		c := node.Consensus.AddPeers(peers)
@@ -275,7 +273,7 @@ func (node *Node) AddPeers(peers []p2p.Peer) int {
 
 func (node *Node) JoinShard(leader p2p.Peer) {
 	// try to join the shard, with 10 minutes time-out
-	backoff := p2p.NewExpBackoff(500*time.Millisecond, 10*time.Minute, 2)
+	backoff := p2p.NewExpBackoff(1*time.Second, 10*time.Minute, 2)
 
 	for node.State == WAIT {
 		backoff.Sleep()
@@ -285,5 +283,4 @@ func (node *Node) JoinShard(leader p2p.Peer) {
 		p2p.SendMessage(leader, buffer)
 		node.log.Debug("Sent ping message")
 	}
-
 }

node/node_handler.go

@@ -528,9 +528,16 @@ func (node *Node) pingMessageHandler(msgPayload []byte) {
 
 	// Send a Pong message back
 	peers := make([]p2p.Peer, 0)
-	for _, v := range node.Neighbors {
-		peers = append(peers, *v)
-	}
+	count := 0
+	node.Neighbors.Range(func(k, v interface{}) bool {
+		if p, ok := v.(p2p.Peer); ok {
+			peers = append(peers, p)
+			count++
+			return true
+		} else {
+			return false
+		}
+	})
 	pong := proto_node.NewPongMessage(peers)
 	buffer := pong.ConstructPongMessage()
 
@@ -547,7 +554,8 @@ func (node *Node) pongMessageHandler(msgPayload []byte) {
 		node.log.Error("Can't get Pong Message")
 		return
 	}
-	node.log.Info("Pong", "Msg", pong)
+	//	node.log.Info("Pong", "Msg", pong)
+	node.State = JOIN
 
 	peers := make([]p2p.Peer, 0)