Merge pull request #264 from ethstorage/bugfix

Fix crashs: add lock ReportPeerSummary and check Peer is null in getIdlePeerForTask

ethstorage/p2p/protocol/peer.go

@@ -5,7 +5,6 @@ package protocol
 
 import (
 	"context"
-	"fmt"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -57,16 +56,6 @@ func (p *Peer) Shards() map[common.Address][]uint64 {
 	return p.shards
 }
 
-// SetShards this should only be set when doing handshake.
-func (p *Peer) SetShards(shards map[common.Address][]uint64) error {
-	// shards can only be set once.
-	if p.shards != nil {
-		return fmt.Errorf("peer shards has been set multi times")
-	}
-	p.shards = shards
-	return nil
-}
-
 // IsShardExist checks whether one specific shard is supported by this peer.
 func (p *Peer) IsShardExist(contract common.Address, shardId uint64) bool {
 	if ids, ok := p.shards[contract]; ok {

ethstorage/p2p/protocol/syncclient.go

@@ -848,8 +848,8 @@ func (s *SyncClient) getIdlePeerForTask(t *task) *Peer {
 		if _, ok := t.statelessPeers[id]; ok {
 			continue
 		}
-		p := s.peers[id]
-		if p.IsShardExist(t.Contract, t.ShardId) {
+		p, ok := s.peers[id]
+		if ok && p.IsShardExist(t.Contract, t.ShardId) {
 			return p
 		}
 	}
@@ -1200,6 +1200,7 @@ func (s *SyncClient) ReportPeerSummary() {
 		select {
 		case <-ticker.C:
 			inbound, outbound := 0, 0
+			s.lock.Lock()
 			for _, p := range s.peers {
 				if p.direction == network.DirInbound {
 					inbound++
@@ -1208,6 +1209,7 @@ func (s *SyncClient) ReportPeerSummary() {
 				}
 			}
 			log.Info("P2P Summary", "activePeers", len(s.peers), "inbound", inbound, "outbound", outbound)
+			s.lock.Unlock()
 		case <-s.resCtx.Done():
 			log.Info("P2P summary stop")
 			return
@@ -1217,6 +1219,9 @@ func (s *SyncClient) ReportPeerSummary() {
 }
 
 func (s *SyncClient) needThisPeer(contractShards map[common.Address][]uint64) bool {
+	if contractShards == nil {
+		return false
+	}
 	for contract, shards := range contractShards {
 		for _, shard := range shards {
 			for _, t := range s.tasks {