Fix global contacts calculation (#868)

* fix global contacts calculation

* fix

* optimizations in download

Author: mateeullahmalik

p2p/kademlia/common.go

@@ -71,6 +71,7 @@ func decompressKeysStr(data []byte) ([]string, error) {
 	return keys, nil
 }
 
+/*
 func compressSymbols(values [][]byte) ([]byte, error) {
 	buf, err := msgpack.Marshal(values)
 	if err != nil {
@@ -97,3 +98,4 @@ func decompressSymbols(data []byte) (values [][]byte, err error) {
 
 	return values, nil
 }
+*/

p2p/kademlia/common_test.go

@@ -120,6 +120,7 @@ func TestCompressAndDecompressKeysStr(t *testing.T) {
 	}
 }
 
+/*
 func TestCompressAndDecompressSymbols(t *testing.T) {
 	tests := []struct {
 		name   string
@@ -156,4 +157,4 @@ func TestCompressAndDecompressSymbols(t *testing.T) {
 			}
 		})
 	}
-}
+}*/

p2p/kademlia/dht.go

@@ -35,6 +35,8 @@ var (
 	delKeysCountThreshold                = 10
 	lowSpaceThreshold                    = 50 // GB
 	batchStoreSize                       = 2500
+	storeSameSymbolsBatchConcurrency     = 1
+	storeSymbolsBatchConcurrency         = 2.0
 )
 
 const maxIterations = 5
@@ -525,7 +527,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 		lenOfKeys = 2000
 	}
 
-	responses := s.batchFindNode(ctx, hashes[:lenOfKeys], knownNodes)
+	responses, _ := s.batchFindNode(ctx, hashes[:lenOfKeys], knownNodes, make(map[string]bool))
 	for response := range responses {
 		if response.Error != nil {
 			log.WithContext(ctx).WithError(response.Error).Error("batch find node failed on a node")
@@ -558,9 +560,6 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 	// remove self from the map
 	delete(knownNodes, string(self.ID))
 
-	// Check if we have enough values locally
-	log.WithContext(ctx).Info("skipping local fetch")
-
 	foundLocalCount, err = s.fetchAndAddLocalKeys(ctx, hexKeys, &resMap, required)
 	if err != nil {
 		return nil, fmt.Errorf("fetch and add local keys: %v", err)
@@ -572,10 +571,10 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
 	}
 
 	// We don't have enough values locally, so we need to fetch from the network
-	batchSize := 2500
+	batchSize := batchStoreSize
 	var networkFound int32
 	totalBatches := int(math.Ceil(float64(required) / float64(batchSize)))
-	parallelBatches := int(math.Min(float64(totalBatches), 2.0))
+	parallelBatches := int(math.Min(float64(totalBatches), storeSymbolsBatchConcurrency))
 
 	semaphore := make(chan struct{}, parallelBatches)
 	var wg sync.WaitGroup
@@ -667,7 +666,7 @@ func (s *DHT) doBatchGetValuesCall(ctx context.Context, node *Node, requestKeys
 
 func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node, keys []string, hexKeys []string, fetchMap map[string][]int,
 	resMap *sync.Map, req, alreadyFound int32) (int, map[string]*NodeList, error) {
-	semaphore := make(chan struct{}, 1) // Limit concurrency to 2
+	semaphore := make(chan struct{}, storeSameSymbolsBatchConcurrency) // Limit concurrency to 1
 	closestContacts := make(map[string]*NodeList)
 	var wg sync.WaitGroup
 	contactsMap := make(map[string]map[string][]*Node)
@@ -1230,6 +1229,7 @@ func (s *DHT) addKnownNodes(nodes []*Node, knownNodes map[string]*Node) {
 func (s *DHT) IterateBatchStore(ctx context.Context, values [][]byte, typ int) error {
 	globalClosestContacts := make(map[string]*NodeList) // This will store the global top 6 nodes for each symbol's hash
 	knownNodes := make(map[string]*Node)                // This will store the nodes we've already contacted
+	contacted := make(map[string]bool)
 	hashes := make([][]byte, len(values))
 
 	for i := 0; i < len(values); i++ {
@@ -1241,47 +1241,73 @@ func (s *DHT) IterateBatchStore(ctx context.Context, values [][]byte, typ int) e
 		s.addKnownNodes(top6.Nodes, knownNodes)
 	}
 
-	localClosestNodes := make(map[string]*NodeList)
-	responses := s.batchFindNode(ctx, hashes, knownNodes)
-	for response := range responses {
-		if response.Error != nil {
-			log.WithContext(ctx).WithError(response.Error).Error("batch find node failed on a node")
-		}
+	var changed bool
+	var i int
+	for {
+		i++
+		changed = false
+		localClosestNodes := make(map[string]*NodeList)
+		responses, atleastOneContacted := s.batchFindNode(ctx, hashes, knownNodes, contacted)
 
-		if response.Message == nil {
-			continue
+		if !atleastOneContacted {
+			break
 		}
 
-		v, ok := response.Message.Data.(*BatchFindNodeResponse)
-		if ok && v.Status.Result == ResultOk {
-			for key, nodesList := range v.ClosestNodes {
-				if nodesList != nil {
-					nl, exists := localClosestNodes[key]
-					if exists {
-						nl.AddNodes(nodesList)
-						localClosestNodes[key] = nl
-					} else {
-						localClosestNodes[key] = &NodeList{Nodes: nodesList, Comparator: base58.Decode(key)}
-					}
+		for response := range responses {
+			if response.Error != nil {
+				log.WithContext(ctx).WithError(response.Error).Error("batch find node failed on a node")
+			}
 
-					s.addKnownNodes(nodesList, knownNodes)
-				}
+			if response.Message == nil {
+				continue
 			}
-		}
-	}
 
-	for key, nodesList := range localClosestNodes {
-		if nodesList == nil {
-			continue
-		}
+			v, ok := response.Message.Data.(*BatchFindNodeResponse)
+			if ok && v.Status.Result == ResultOk {
+				for key, nodesList := range v.ClosestNodes {
+					if nodesList != nil {
+						nl, exists := localClosestNodes[key]
+						if exists {
+							nl.AddNodes(nodesList)
+							localClosestNodes[key] = nl
+						} else {
+							localClosestNodes[key] = &NodeList{Nodes: nodesList, Comparator: base58.Decode(key)}
+						}
 
-		nodesList.Comparator = base58.Decode(key)
-		nodesList.Sort()
-		nodesList.TopN(Alpha)
+						s.addKnownNodes(nodesList, knownNodes)
+					}
+				}
+			}
+		}
 
 		// we now need to check if the nodes in the globalClosestContacts Map are still in the top 6
 		// if yes, we can store the data to them
 		// if not, we need to send calls to the newly found nodes to inquire about the top 6 nodes
+		for key, nodesList := range localClosestNodes {
+			if nodesList == nil {
+				continue
+			}
+
+			nodesList.Comparator = base58.Decode(key)
+			nodesList.Sort()
+			nodesList.TopN(Alpha)
+			s.addKnownNodes(nodesList.Nodes, knownNodes)
+
+			if !haveAllNodes(nodesList.Nodes, globalClosestContacts[key].Nodes) {
+				log.WithContext(ctx).WithField("key", key).WithField("have", nodesList.String()).WithField("got", globalClosestContacts[key].String()).Info("global closest contacts list changed!")
+				changed = true
+			}
+
+			nodesList.AddNodes(globalClosestContacts[key].Nodes)
+			nodesList.Sort()
+			nodesList.TopN(Alpha)
+			globalClosestContacts[key] = nodesList
+		}
+
+		if !changed {
+			log.WithContext(ctx).WithField("iter", i).Info("global closest contacts list did not change, we can now store the data")
+			break
+		}
 	}
 
 	// assume at this point, we have True\Golabl top 6 nodes for each symbol's hash stored in globalClosestContacts Map
@@ -1356,20 +1382,15 @@ func (s *DHT) batchStoreNetwork(ctx context.Context, values [][]byte, nodes map[
 			default:
 				keysToStore := storageMap[key]
 				toStore := make([][]byte, len(keysToStore))
+				totalBytes := 0
 				for i, idx := range keysToStore {
 					toStore[i] = values[idx]
+					totalBytes += len(values[idx])
 				}
 
-				payload, err := compressSymbols(toStore)
-				if err != nil {
-					log.P2P().WithContext(ctx).WithError(err).Error("compress symbols failed")
-					responses <- &MessageWithError{Error: err}
-					return
-				}
-
-				log.WithContext(ctx).WithField("keys", len(toStore)).WithField("data-size", utils.BytesIntToMB(len(payload))).Info("batch store to node payload size")
+				log.WithContext(ctx).WithField("keys", len(toStore)).WithField("size-before-compress", utils.BytesIntToMB(totalBytes)).Info("batch store to node")
 
-				data := &BatchStoreDataRequest{Data: payload, Type: typ}
+				data := &BatchStoreDataRequest{Data: toStore, Type: typ}
 				request := s.newMessage(BatchStoreData, receiver, data)
 				response, err := s.network.Call(ctx, request, true)
 				if err != nil {
@@ -1389,12 +1410,18 @@ func (s *DHT) batchStoreNetwork(ctx context.Context, values [][]byte, nodes map[
 	return responses
 }
 
-func (s *DHT) batchFindNode(ctx context.Context, payload [][]byte, nodes map[string]*Node) chan *MessageWithError {
+func (s *DHT) batchFindNode(ctx context.Context, payload [][]byte, nodes map[string]*Node, contacted map[string]bool) (chan *MessageWithError, bool) {
 	responses := make(chan *MessageWithError, len(nodes))
-
+	atleastOneContacted := false
 	var wg sync.WaitGroup
 
 	for _, node := range nodes {
+		if _, ok := contacted[string(node.ID)]; ok {
+			continue
+		}
+
+		contacted[string(node.ID)] = true
+		atleastOneContacted = true
 		wg.Add(1)
 		go func(receiver *Node) {
 			defer wg.Done()
@@ -1420,5 +1447,5 @@ func (s *DHT) batchFindNode(ctx context.Context, payload [][]byte, nodes map[str
 	wg.Wait()
 	close(responses)
 
-	return responses
+	return responses, atleastOneContacted
 }

p2p/kademlia/message.go

@@ -219,7 +219,7 @@ func decode(conn io.Reader) (*Message, error) {
 
 // BatchStoreDataRequest defines the request data for store data
 type BatchStoreDataRequest struct {
-	Data []byte
+	Data [][]byte
 	Type int
 }
 

p2p/kademlia/network.go

@@ -843,13 +843,7 @@ func (s *Network) handleBatchStoreData(ctx context.Context, message *Message) (r
 	// add the sender to queries hash table
 	s.dht.addNode(ctx, message.Sender)
 
-	values, err := decompressSymbols(request.Data)
-	if err != nil {
-		err = errors.Errorf("batchStore: decompress symbols: %w", err)
-		return s.generateResponseMessage(BatchStoreData, message.Sender, ResultFailed, err.Error())
-	}
-
-	if err := s.dht.store.StoreBatch(ctx, values, 1, false); err != nil {
+	if err := s.dht.store.StoreBatch(ctx, request.Data, 1, false); err != nil {
 		err = errors.Errorf("batch store the data: %w", err)
 		return s.generateResponseMessage(BatchStoreData, message.Sender, ResultFailed, err.Error())
 	}

p2p/kademlia/node.go

@@ -78,6 +78,23 @@ func (s *NodeList) DelNode(node *Node) {
 	}
 }
 
+func haveAllNodes(nodesA []*Node, nodesB []*Node) bool {
+	for _, nodeA := range nodesA {
+		found := false
+		for _, nodeB := range nodesB {
+			if bytes.Equal(nodeA.ID, nodeB.ID) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false
+		}
+	}
+
+	return true
+}
+
 // Exists return true if the node is already there
 func (s *NodeList) Exists(node *Node) bool {
 	s.Mux.RLock()

supernode/services/common/storage_handler.go

@@ -21,7 +21,7 @@ import (
 )
 
 const (
-	loadSymbolsBatchSize = 3000
+	loadSymbolsBatchSize = 2500
 	storeSymbolsPercent  = 10
 )