Merge pull request #650 from Arceliar/search

More search updates

src/yggdrasil/search.go

@@ -27,6 +27,7 @@ const search_MAX_SEARCH_SIZE = 16
 
 // This defines the time after which we time out a search (so it can restart).
 const search_RETRY_TIME = 3 * time.Second
+const search_STEP_TIME = 100 * time.Millisecond
 
 // Information about an ongoing search.
 // Includes the target NodeID, the bitmask to match it to an IP, and the list of nodes to visit / already visited.
@@ -78,34 +79,39 @@ func (s *searches) createSearch(dest *crypto.NodeID, mask *crypto.NodeID, callba
 // If there is, it adds the response info to the search and triggers a new search step.
 // If there's no ongoing search, or we if the dhtRes finished the search (it was from the target node), then don't do anything more.
 func (sinfo *searchInfo) handleDHTRes(res *dhtRes) {
+	old := sinfo.visited
+	if res != nil {
 		sinfo.recv++
-	if res == nil || sinfo.checkDHTRes(res) {
-		// Either we don't recognize this search, or we just finished it
-		return
+		if sinfo.checkDHTRes(res) {
+			return // Search finished successfully
 		}
-	// Add to the search and continue
+		// Add results to the search
 		sinfo.addToSearch(res)
+	}
+	if res == nil || sinfo.visited != old {
+		// Continue the search
 		sinfo.doSearchStep()
+	}
 }
 
 // Adds the information from a dhtRes to an ongoing search.
 // Info about a node that has already been visited is not re-added to the search.
-// Duplicate information about nodes toVisit is deduplicated (the newest information is kept).
 func (sinfo *searchInfo) addToSearch(res *dhtRes) {
+	// Add to search
 	for _, info := range res.Infos {
-		if *info.getNodeID() == sinfo.visited {
-			// dht_ordered could return true here, but we want to skip it in this case
-			continue
-		}
-		if dht_ordered(&sinfo.dest, info.getNodeID(), &sinfo.visited) && *info.getNodeID() != sinfo.visited {
-			// Response is closer to the destination
 		sinfo.toVisit = append(sinfo.toVisit, info)
 	}
 	// Sort
 	sort.SliceStable(sinfo.toVisit, func(i, j int) bool {
 		// Should return true if i is closer to the destination than j
-			return dht_ordered(&res.Dest, sinfo.toVisit[i].getNodeID(), sinfo.toVisit[j].getNodeID())
+		return dht_ordered(&sinfo.dest, sinfo.toVisit[i].getNodeID(), sinfo.toVisit[j].getNodeID())
 	})
+	// Remove anything too far away
+	for idx, info := range sinfo.toVisit {
+		if *info.getNodeID() == sinfo.visited || !dht_ordered(&sinfo.dest, info.getNodeID(), &sinfo.visited) {
+			sinfo.toVisit = sinfo.toVisit[:idx]
+			break
+		}
 	}
 }
 
@@ -121,14 +127,14 @@ func (sinfo *searchInfo) doSearchStep() {
 		return
 	}
 	// Send to the next search target
-	for _, next := range sinfo.toVisit {
+	if len(sinfo.toVisit) > 0 {
+		next := sinfo.toVisit[0]
+		sinfo.toVisit = sinfo.toVisit[1:]
 		rq := dhtReqKey{next.key, sinfo.dest}
 		sinfo.searches.router.dht.addCallback(&rq, sinfo.handleDHTRes)
 		sinfo.searches.router.dht.ping(next, &sinfo.dest)
-		sinfo.time = time.Now()
 		sinfo.send++
 	}
-	sinfo.toVisit = sinfo.toVisit[:0]
 }
 
 // If we've recently sent a ping for this search, do nothing.
@@ -138,7 +144,7 @@ func (sinfo *searchInfo) continueSearch() {
 	// In case the search dies, try to spawn another thread later
 	// Note that this will spawn multiple parallel searches as time passes
 	// Any that die aren't restarted, but a new one will start later
-	time.AfterFunc(search_RETRY_TIME, func() {
+	time.AfterFunc(search_STEP_TIME, func() {
 		sinfo.searches.router.Act(nil, func() {
 			// FIXME this keeps the search alive forever if not for the searches map, fix that
 			newSearchInfo := sinfo.searches.searches[sinfo.dest]
@@ -167,10 +173,11 @@ func (s *searches) newIterSearch(dest *crypto.NodeID, mask *crypto.NodeID, callb
 // Otherwise return false.
 func (sinfo *searchInfo) checkDHTRes(res *dhtRes) bool {
 	from := dhtInfo{key: res.Key, coords: res.Coords}
-	if dht_ordered(&sinfo.dest, from.getNodeID(), &sinfo.visited) {
+	if *from.getNodeID() != sinfo.visited && dht_ordered(&sinfo.dest, from.getNodeID(), &sinfo.visited) {
 		// Closer to the destination, so update visited
 		sinfo.searches.router.core.log.Debugln("Updating search:", sinfo.dest, *from.getNodeID(), sinfo.send, sinfo.recv)
 		sinfo.visited = *from.getNodeID()
+		sinfo.time = time.Now()
 	}
 	them := from.getNodeID()
 	var destMasked crypto.NodeID
@@ -194,8 +201,11 @@ func (sinfo *searchInfo) checkDHTRes(res *dhtRes) bool {
 			sinfo.callback(sess, nil)
 		}
 		// Cleanup
+		if _, isIn := sinfo.searches.searches[sinfo.dest]; isIn {
+			sinfo.searches.router.core.log.Debugln("Finished search:", sinfo.dest, sinfo.send, sinfo.recv)
 			delete(sinfo.searches.searches, res.Dest)
 		}
+	}
 	// They match, so create a session and send a sessionRequest
 	var err error
 	sess, isIn := sinfo.searches.router.sessions.getByTheirPerm(&res.Key)