diff --git a/src/yggdrasil/search.go b/src/yggdrasil/search.go index 6bd53c6..ede4547 100644 --- a/src/yggdrasil/search.go +++ b/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) { - sinfo.recv++ - if res == nil || sinfo.checkDHTRes(res) { - // Either we don't recognize this search, or we just finished it - return + old := sinfo.visited + if res != nil { + sinfo.recv++ + if sinfo.checkDHTRes(res) { + return // Search finished successfully + } + // Add results to the search + sinfo.addToSearch(res) + } + if res == nil || sinfo.visited != old { + // Continue the search + sinfo.doSearchStep() } - // Add to the search and continue - sinfo.addToSearch(res) - 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 + 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(&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 } - 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()) - }) } } @@ -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,7 +201,10 @@ func (sinfo *searchInfo) checkDHTRes(res *dhtRes) bool { sinfo.callback(sess, nil) } // Cleanup - delete(sinfo.searches.searches, res.Dest) + 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