Merge pull request #643 from pastelnetwork/PSL-942_p2pFixes

fix sort based on xor, fix check node activity worker

Author: mateeullahmalik

p2p/client.go

@@ -29,4 +29,10 @@ type Client interface {
 	// LocalStore store data to the network, which will trigger the iterative store message
 	// - the base58 encoded identifier will be returned
 	LocalStore(ctx context.Context, key string, data []byte) (string, error)
+
+	// DisableKey adds key to disabled keys list - It takes in a B58 encoded SHA-256 hash
+	DisableKey(ctx context.Context, b58EncodedHash string) error
+
+	// EnableKey removes key from disabled list - It takes in a B58 encoded SHA-256 hash
+	EnableKey(ctx context.Context, b58EncodedHash string) error
 }

p2p/kademlia/banlist.go

@@ -13,7 +13,7 @@ const (
 	banDuration = 3 * time.Hour
 
 	// threshold - threshold
-	threshold = 5
+	threshold = 2
 )
 
 // BanNode is the over-the-wire representation of a node
@@ -166,7 +166,9 @@ func (s *BanList) ToNodeList() []*Node {
 	ret := make([]*Node, 0)
 
 	for i := 0; i < len(s.Nodes); i++ {
-		ret = append(ret, &s.Nodes[i].Node)
+		if s.Nodes[i].count > threshold {
+			ret = append(ret, &s.Nodes[i].Node)
+		}
 	}
 
 	return ret

p2p/kademlia/dht.go

@@ -18,7 +18,6 @@ import (
 	"github.com/pastelnetwork/gonode/common/storage"
 	"github.com/pastelnetwork/gonode/common/storage/memory"
 	"github.com/pastelnetwork/gonode/common/utils"
-	"github.com/pastelnetwork/gonode/p2p/kademlia/domain"
 	"github.com/pastelnetwork/gonode/pastel"
 	"golang.org/x/crypto/sha3"
 )
@@ -36,20 +35,19 @@ const maxIterations = 5
 
 // DHT represents the state of the local node in the distributed hash table
 type DHT struct {
-	ht                   *HashTable       // the hashtable for routing
-	options              *Options         // the options of DHT
-	network              *Network         // the network of DHT
-	store                Store            // the storage of DHT
-	metaStore            MetaStore        // the meta storage of DHT
-	done                 chan struct{}    // distributed hash table is done
-	cache                storage.KeyValue // store bad bootstrap addresses
-	pastelClient         pastel.Client
-	externalIP           string
-	mtx                  sync.Mutex
-	authHelper           *AuthHelper
-	ignorelist           *BanList
-	nodeReplicationTimes map[string]domain.NodeReplicationInfo
-	replicationMtx       sync.RWMutex
+	ht             *HashTable       // the hashtable for routing
+	options        *Options         // the options of DHT
+	network        *Network         // the network of DHT
+	store          Store            // the storage of DHT
+	metaStore      MetaStore        // the meta storage of DHT
+	done           chan struct{}    // distributed hash table is done
+	cache          storage.KeyValue // store bad bootstrap addresses
+	pastelClient   pastel.Client
+	externalIP     string
+	mtx            sync.Mutex
+	authHelper     *AuthHelper
+	ignorelist     *BanList
+	replicationMtx sync.RWMutex
 }
 
 // Options contains configuration options for the local node
@@ -84,26 +82,16 @@ func NewDHT(ctx context.Context, store Store, metaStore MetaStore, pc pastel.Cli
 	if options.Port <= 0 {
 		options.Port = defaultNetworkPort
 	}
-	info, err := store.GetAllReplicationInfo(ctx)
-	if err != nil {
-		log.P2P().WithContext(ctx).WithError(err).Errorf("get all replicationInfo failed")
-	}
-
-	replicationMap := make(map[string]domain.NodeReplicationInfo)
-	for _, v := range info {
-		replicationMap[string(v.ID)] = v
-	}
 
 	s := &DHT{
-		metaStore:            metaStore,
-		store:                store,
-		options:              options,
-		pastelClient:         pc,
-		done:                 make(chan struct{}),
-		cache:                memory.NewKeyValue(),
-		ignorelist:           NewBanList(ctx),
-		nodeReplicationTimes: replicationMap,
-		replicationMtx:       sync.RWMutex{},
+		metaStore:      metaStore,
+		store:          store,
+		options:        options,
+		pastelClient:   pc,
+		done:           make(chan struct{}),
+		cache:          memory.NewKeyValue(),
+		ignorelist:     NewBanList(ctx),
+		replicationMtx: sync.RWMutex{},
 	}
 
 	if options.ExternalIP != "" {
@@ -482,7 +470,7 @@ func (s *DHT) iterate(ctx context.Context, iterativeType int, target []byte, dat
 
 	igList := s.ignorelist.ToNodeList()
 	// find the closest contacts for the target node from local route tables
-	nl := s.ht.closestContacts(Alpha, target, igList)
+	nl, _ := s.ht.closestContacts(Alpha, target, igList)
 	if len(igList) > 0 {
 		log.P2P().WithContext(ctx).WithField("nodes", nl.String()).WithField("ignored", s.ignorelist.String()).Info("closest contacts")
 	}
@@ -633,7 +621,7 @@ func (s *DHT) iterateFindValue(ctx context.Context, iterativeType int, target []
 	igList := s.ignorelist.ToNodeList()
 
 	// nl will have the closest nodes to the target value, it will ignore the nodes in igList
-	nl := s.ht.closestContacts(Alpha, target, igList)
+	nl, _ := s.ht.closestContacts(Alpha, target, igList)
 	if len(igList) > 0 {
 		log.P2P().WithContext(ctx).WithField("nodes", nl.String()).WithField("ignored", s.ignorelist.String()).Info("closest contacts")
 	}
@@ -785,7 +773,7 @@ func (s *DHT) addNode(ctx context.Context, node *Node) *Node {
 func (s *DHT) NClosestNodes(_ context.Context, n int, key string, ignores ...*Node) []*Node {
 	list := s.ignorelist.ToNodeList()
 	ignores = append(ignores, list...)
-	nodeList := s.ht.closestContacts(n, base58.Decode(key), ignores)
+	nodeList, _ := s.ht.closestContacts(n, base58.Decode(key), ignores)
 
 	return nodeList.Nodes
 }

p2p/kademlia/fetch_and_store.go

@@ -118,7 +118,7 @@ func (s *DHT) BatchFetchAndStoreFailedKeys(ctx context.Context) error {
 	repKeys := make([]domain.ToRepKey, 0, len(keys))
 	for i := 0; i < len(keys); i++ {
 		igList := s.ignorelist.ToNodeList()
-		nl := s.ht.closestContacts(failedKeysClosestContactsLookupCount, keys[i].Key, igList)
+		nl, _ := s.ht.closestContacts(failedKeysClosestContactsLookupCount, keys[i].Key, igList)
 		attempt := (keys[i].Attempts - maxBatchAttempts) + 1
 
 		if len(nl.Nodes) > attempt {

p2p/kademlia/hashtable.go

@@ -189,7 +189,7 @@ func (ht *HashTable) hasNode(id []byte) bool {
 }
 
 // closestContacts returns the closest contacts of target
-func (ht *HashTable) closestContacts(num int, target []byte, ignoredNodes []*Node) *NodeList {
+func (ht *HashTable) closestContacts(num int, target []byte, ignoredNodes []*Node) (*NodeList, int) {
 	ht.mutex.RLock()
 	defer ht.mutex.RUnlock()
 
@@ -203,9 +203,11 @@ func (ht *HashTable) closestContacts(num int, target []byte, ignoredNodes []*Nod
 		Comparator: target,
 	}
 
+	counter := 0
 	// Flatten the routeTable and add nodes to nl if they're not in the ignoredMap
 	for _, bucket := range ht.routeTable {
 		for _, node := range bucket {
+			counter++
 			if !ignoredMap[string(node.ID)] {
 				nl.AddNodes([]*Node{node})
 			}
@@ -216,7 +218,7 @@ func (ht *HashTable) closestContacts(num int, target []byte, ignoredNodes []*Nod
 	nl.Sort()
 	nl.TopN(num)
 
-	return nl
+	return nl, counter
 }
 
 // bucketIndex return the bucket index from two node ids
@@ -298,9 +300,11 @@ func (ht *HashTable) closestContactsWithInlcudingNode(num int, target []byte, ig
 	}
 
 	// Flatten the routeTable and add nodes to nl if they're not in the ignoredMap
+	counter := 0
 	for _, bucket := range ht.routeTable {
 		for _, node := range bucket {
 			if !ignoredMap[string(node.ID)] {
+				counter++
 				nl.AddNodes([]*Node{node})
 			}
 		}

p2p/kademlia/network.go

@@ -153,7 +153,7 @@ func (s *Network) handleFindNode(ctx context.Context, message *Message) (res []b
 
 	// the closest contacts
 	hashedTargetID, _ := utils.Sha3256hash(request.Target)
-	closest := s.dht.ht.closestContacts(K, hashedTargetID, []*Node{message.Sender})
+	closest, _ := s.dht.ht.closestContacts(K, hashedTargetID, []*Node{message.Sender})
 
 	response := &FindNodeResponse{
 		Status: ResponseStatus{
@@ -227,7 +227,7 @@ func (s *Network) handleFindValue(ctx context.Context, message *Message) (res []
 			},
 		}
 
-		closest := s.dht.ht.closestContacts(K, request.Target, []*Node{message.Sender})
+		closest, _ := s.dht.ht.closestContacts(K, request.Target, []*Node{message.Sender})
 		response.Closest = closest.Nodes
 
 		// new a response message
@@ -246,7 +246,7 @@ func (s *Network) handleFindValue(ctx context.Context, message *Message) (res []
 		response.Value = value
 	} else {
 		// return the closest contacts
-		closest := s.dht.ht.closestContacts(K, request.Target, []*Node{message.Sender})
+		closest, _ := s.dht.ht.closestContacts(K, request.Target, []*Node{message.Sender})
 		response.Closest = closest.Nodes
 	}
 

p2p/kademlia/node.go

@@ -7,10 +7,6 @@ import (
 	"sort"
 	"strings"
 	"sync"
-
-	"github.com/btcsuite/btcutil/base58"
-	"github.com/pastelnetwork/gonode/common/log"
-	"github.com/pastelnetwork/gonode/common/utils"
 )
 
 // Node is the over-the-wire representation of a node
@@ -26,7 +22,7 @@ type Node struct {
 }
 
 func (s *Node) String() string {
-	return fmt.Sprintf("%v-%v:%d", base58.Encode(s.ID), s.IP, s.Port)
+	return fmt.Sprintf("%v-%v:%d", string(s.ID), s.IP, s.Port)
 }
 
 // NodeList is used in order to sort a list of nodes
@@ -81,8 +77,8 @@ func (s *NodeList) Exists(node *Node) bool {
 }
 
 func (s *NodeList) exists(node *Node) bool {
-	for _, item := range s.Nodes {
-		if bytes.Equal(item.ID, node.ID) {
+	for i := 0; i < len(s.Nodes); i++ {
+		if bytes.Equal(s.Nodes[i].ID, node.ID) {
 			return true
 		}
 	}
@@ -106,13 +102,6 @@ func (s *NodeList) Len() int {
 	return len(s.Nodes)
 }
 
-func (s *NodeList) distance(id1, id2 []byte) *big.Int {
-	o1 := new(big.Int).SetBytes(id1)
-	o2 := new(big.Int).SetBytes(id2)
-
-	return new(big.Int).Xor(o1, o2)
-}
-
 // AddFirst adds a node to the first position of the list.
 func (s *NodeList) AddFirst(node *Node) {
 	s.Mux.Lock()         // lock for writing
@@ -134,32 +123,20 @@ func (s *NodeList) TopN(n int) {
 	}
 }
 
-func (s *NodeList) computeDistances() []big.Int {
-	distances := make([]big.Int, s.Len())
-	for i, node := range s.Nodes {
-		cID, _ := utils.Sha3256hash(node.ID)
-		dist := s.distance(cID, s.Comparator)
-		distances[i] = *dist
+func (s *NodeList) distance(id1, id2 []byte) *big.Int {
+	o1 := new(big.Int).SetBytes(id1)
+	o2 := new(big.Int).SetBytes(id2)
 
-		if s.debug {
-			log.WithField("node", node.String()).WithField("dist", distances[i]).Info("computeDistances")
-		}
-	}
-	return distances
+	return new(big.Int).Xor(o1, o2)
 }
 
 // Sort sorts nodes
 func (s *NodeList) Sort() {
 	s.Mux.Lock()
 	defer s.Mux.Unlock()
 
-	// Compute distances
-	distances := s.computeDistances()
-
 	// Sort using the precomputed distances
-	sort.Slice(s.Nodes, func(i, j int) bool {
-		return distances[i].Cmp(&distances[j]) == -1
-	})
+	sort.Sort(s)
 }
 
 // Swap swap two nodes